pain001.logging_schema package

Standardized logging schema for Pain001.

This package provides a centralized logging structure for consistent, machine-parsable log output across CLI and library components. All log entries are JSON-formatted for easy integration with log aggregation systems (Elasticsearch, Splunk, CloudWatch, etc.).

The implementation is split across focused submodules; this package re-exports the full public surface, so from pain001.logging_schema import ... continues to work unchanged:

• schema — log level, status, event, and field name vocabularies (LogLevel, ExecutionStatus, Events, Fields).

• context — per-request id propagation.

• redaction — PII masking and log-injection sanitization.

• events — structured event emitters (log_event and friends).

• tracker — ExecutionSummaryTracker.

• formatter — JSONFormatter and configure_json_logging.

• metrics — ExecutionMetrics.

IMPORTANT: PII Protection

This module implements automatic PII redaction for sensitive fields. Any field containing IBAN, BIC, or personal names is automatically masked before logging to ensure GDPR/PCI-DSS compliance.

Request Tracing

Every operation is assigned a unique request_id (UUID) to enable end-to-end request tracking across distributed systems and microservices. This is essential for API Layer (#149) observability.

Log Severity Mapping (ISO 20022 Context)

• DEBUG: XSD traversal, template loading, variable substitution

• INFO: Process start/success, validation success, file generation

• WARNING: Schema deprecation, character truncation, missing optional fields

• ERROR: XSD validation failure, checksum failure, bank profile violations

• CRITICAL: Missing dependencies, memory overflow, configuration corruption

Event Naming Convention:

• Use snake_case for event names

• Format: <component>_<action>_<state>

• Examples: “process_start”, “validation_success”, “xml_generated”

Field Naming Convention:

• Use snake_case for field names

• Be consistent with terminology across all events

• Include units where applicable (e.g., “duration_ms”, “size_bytes”)

• All logs are flat JSON objects for easy indexing

Submodules

pain001.logging_schema.context module

Per-request ID propagation for distributed tracing.

pain001.logging_schema.context.generate_request_id() → str

Generate a unique request ID for request tracing.

Returns:

req-<8-char-hex>).

Return type:

A short UUID-based request ID (format

Example

>>> generate_request_id()
'req-88f24b21'

pain001.logging_schema.context.get_request_id() → str

Get or create request ID for current context.

Returns:

The request ID for the current execution context.

pain001.logging_schema.context.set_request_id(request_id: str) → None

Set request ID for current context (useful for API handlers).

Parameters:

request_id – The request ID to set.

pain001.logging_schema.events module

Structured event emitters that build flat, PII-redacted JSON log records.

pain001.logging_schema.events.log_data_load_event(logger: Logger, data_source_type: str, success: bool, record_count: int | None = None, error: Exception | None = None, duration_ms: int | None = None) → None

Log data loading event.

Parameters:

• logger – The logger instance to use.

• data_source_type – Type of data source (csv, sqlite, list, dict).

• success – Whether data loading succeeded.

• record_count – Number of records loaded (if success).

• error – Exception if loading failed (None if success).

• duration_ms – Loading duration in milliseconds.

pain001.logging_schema.events.log_event(logger: Logger, level: int, event: str, **fields: Any) → None

Log a structured event with standardized format and PII redaction.

This function automatically: 1. Adds request_id for distributed tracing 2. Adds ISO 8601 timestamp 3. Redacts PII before logging (GDPR/PCI-DSS compliance) 4. Outputs flat JSON for easy indexing

Parameters:

• logger – The logger instance to use.

• level – Logging level (logging.INFO, logging.ERROR, etc.).

• event – Event name from Events class.

• **fields – Additional fields to include in the log entry.

Example

>>> log_event(
...     logger,
...     logging.INFO,
...     Events.PROCESS_START,
...     message_type="pain.001.001.03",
...     record_count=10
... )
# Output: {"timestamp": "2026-01-14T21:59:55Z", "level": "INFO",
#          "request_id": "req-88f24b21", "event": "process_start", ...}

pain001.logging_schema.events.log_process_error(logger: Logger, error: Exception, message_type: str | None = None, **extra_fields: Any) → None

Log process error event.

Parameters:

• logger – The logger instance to use.

• error – The exception that occurred.

• message_type – ISO 20022 message type (if known).

• **extra_fields – Additional fields to include.

pain001.logging_schema.events.log_process_start(logger: Logger, message_type: str, data_source_type: str, **extra_fields: Any) → float

Log process start event and return start timestamp.

Parameters:

• logger – The logger instance to use.

• message_type – ISO 20022 message type.

• data_source_type – Type of data source (csv, sqlite, list, dict).

• **extra_fields – Additional fields to include.

Returns:

Start timestamp for duration calculation.

pain001.logging_schema.events.log_process_success(logger: Logger, start_time: float, message_type: str, record_count: int, **extra_fields: Any) → None

Log process success event with duration.

Parameters:

• logger – The logger instance to use.

• start_time – Start timestamp from log_process_start().

• message_type – ISO 20022 message type.

• record_count – Number of records processed.

• **extra_fields – Additional fields to include.

pain001.logging_schema.events.log_validation_event(logger: Logger, validation_type: str, success: bool, error: Exception | None = None, **extra_fields: Any) → None

Log validation event (success or error).

Parameters:

• logger – The logger instance to use.

• validation_type – Type of validation (schema, data, business_rules).

• success – Whether validation succeeded.

• error – Exception if validation failed (None if success).

• **extra_fields – Additional fields to include.

pain001.logging_schema.events.log_xml_generation_event(logger: Logger, message_type: str, success: bool, record_count: int | None = None, error: Exception | None = None, duration_ms: int | None = None) → None

Log XML generation event.

Parameters:

• logger – The logger instance to use.

• message_type – ISO 20022 message type.

• success – Whether XML generation succeeded.

• record_count – Number of records in generated XML.

• error – Exception if generation failed (None if success).

• duration_ms – Generation duration in milliseconds.

pain001.logging_schema.formatter module

JSON log formatter and handler configuration (Issue #149).

class pain001.logging_schema.formatter.JSONFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

Custom JSON formatter for structured logging output.

This formatter ensures all log records are emitted as valid JSON, regardless of the logging method used (logger.info(), logger.error(), etc.). It automatically adds standard fields (timestamp, level, logger name, request_id) and merges them with structured log_event() calls.

Example

>>> handler = logging.StreamHandler()
>>> handler.setFormatter(JSONFormatter())
>>> logger.addHandler(handler)

format(record: LogRecord) → str

Format log record as JSON.

Parameters:

record – LogRecord to format.

Returns:

JSON-formatted log entry as string.

pain001.logging_schema.formatter.configure_json_logging(logger: Logger | None = None, level: str | int = 20, log_file: str | None = None, max_bytes: int = 10485760, backup_count: int = 5, console_output: bool = True) → Logger

Configure structured JSON logging for Pain001.

This function sets up production-ready JSON logging with: - JSON formatter for all handlers - Optional file rotation (for persistent logs) - Console output (for containerized environments) - PII redaction (automatic via log_event()) - Request ID tracing

Environment Variables:

PAIN001_LOG_LEVEL: Override log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) PAIN001_LOG_FILE: Override log file path PAIN001_LOG_JSON: Enable JSON logging (true/false)

Parameters:

• logger – Logger to configure (defaults to root logger).

• level – Minimum log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).

• log_file – Path to log file (None = console only).

• max_bytes – Max file size before rotation (default: 10MB).

• backup_count – Number of backup files to keep (default: 5).

• console_output – Whether to log to console (default: True).

Returns:

Configured logger instance.

Example

>>> # Simple console logging
>>> logger = configure_json_logging()
>>> log_event(logger, logging.INFO, Events.PROCESS_START)

>>> # Production setup with file rotation
>>> logger = configure_json_logging(
...     log_file="/var/log/pain001/app.log",
...     level=logging.INFO,
...     max_bytes=50*1024*1024,  # 50MB
...     backup_count=10
... )

>>> # Docker/Kubernetes setup (console only)
>>> logger = configure_json_logging(console_output=True)

pain001.logging_schema.metrics module

Detailed execution telemetry for API observability and tracing.

class pain001.logging_schema.metrics.ExecutionMetrics(logger: Logger, operation: str, message_type: str | None = None, request_id: str | None = None)

Bases: object

Enhanced execution metrics tracking with detailed telemetry.

This class extends ExecutionSummaryTracker with additional metrics for API observability, performance monitoring, and distributed tracing. Tracks detailed timing breakdowns, resource usage, and validation results.

Parameters:

• logger – Logger instance for telemetry output.

• operation – Operation being tracked (e.g., “xml_generation”).

• message_type – ISO 20022 message type (if applicable).

• request_id – Request ID for distributed tracing (auto-generated if None).

Example

>>> metrics = ExecutionMetrics(
...     logger=logger,
...     operation="xml_generation",
...     message_type="pain.001.001.03"
... )
>>> metrics.start()
>>> metrics.track_phase("data_load", duration_ms=120)
>>> metrics.track_phase("xml_generation", duration_ms=350)
>>> metrics.track_validation("schema", "PASSED")
>>> metrics.log_telemetry()

increment_failed(count: int = 1) → None

Increment failed record count.

Parameters:

count – Number of failed records to add (default: 1).

increment_processed(count: int = 1) → None

Increment processed record count.

Parameters:

count – Number of records to add (default: 1).

log_telemetry() → None

Log comprehensive telemetry report.

set_error(error_message: str) → None

Set error message and mark as failed.

Parameters:

error_message – Error description.

start() → None

Mark operation start time.

track_phase(phase_name: str, duration_ms: int) → None

Track timing for a specific phase.

Parameters:

• phase_name – Name of the phase (e.g., “data_load”, “xml_generation”).

• duration_ms – Duration in milliseconds.

track_validation(validation_type: str, status: str) → None

Track validation result.

Parameters:

• validation_type – Type of validation (e.g., “schema”, “business_rules”).

• status – Result status (e.g., “PASSED”, “FAILED”).

pain001.logging_schema.redaction module

PII redaction and log-injection sanitization (GDPR/PCI-DSS, CWE-117).

pain001.logging_schema.redaction.mask_sensitive_data(value: str, visible_chars: int = 4) → str

Mask sensitive data for logging.

Parameters:

• value – The sensitive value to mask.

• visible_chars – Number of characters to show at start and end.

Returns:

Masked string showing only first and last visible_chars.

Examples

>>> mask_sensitive_data("GB29NWBK60161331926819", 4)
'GB29****6819'
>>> mask_sensitive_data("Short", 4)
'****'

pain001.logging_schema.schema module

Constant vocabularies for structured logging (levels, statuses, events, fields).

These classes are namespaces of string constants rather than enums so they serialize directly to JSON and read naturally at call sites.

class pain001.logging_schema.schema.Events

Bases: object

Standardized event names for structured logging.

CLI_ARGS_PARSED = 'cli_args_parsed'

CLI_DRY_RUN = 'cli_dry_run'

DATA_LOAD_ERROR = 'data_load_error'

DATA_LOAD_START = 'data_load_start'

DATA_LOAD_SUCCESS = 'data_load_success'

EXECUTION_SUMMARY = 'execution_summary'

NAMESPACE_REGISTER = 'namespace_register'

PROCESS_ERROR = 'process_error'

PROCESS_START = 'process_start'

PROCESS_SUCCESS = 'process_success'

VALIDATION_ERROR = 'validation_error'

VALIDATION_START = 'validation_start'

VALIDATION_SUCCESS = 'validation_success'

XML_GENERATE_ERROR = 'xml_generate_error'

XML_GENERATE_START = 'xml_generate_start'

XML_GENERATE_SUCCESS = 'xml_generate_success'

XSD_VALIDATION_ERROR = 'xsd_validation_error'

XSD_VALIDATION_START = 'xsd_validation_start'

XSD_VALIDATION_SUCCESS = 'xsd_validation_success'

class pain001.logging_schema.schema.ExecutionStatus

Bases: object

High-level execution status for summary reports.

ABORTED = 'ABORTED'

COMPLETED_WITH_WARNINGS = 'COMPLETED_WITH_WARNINGS'

FAILED = 'FAILED'

SUCCESS = 'SUCCESS'

class pain001.logging_schema.schema.Fields

Bases: object

Standardized field names for structured logging.

BANK_PROFILE = 'bank_profile'

COMPONENT = 'component'

DATA_SOURCE_TYPE = 'data_source_type'

DRY_RUN = 'dry_run'

DURATION_MS = 'duration_ms'

END_TO_END_ID = 'end_to_end_id'

ERROR_FIELD = 'error_field'

ERROR_INVALID_VALUE = 'error_invalid_value'

ERROR_MESSAGE = 'error_message'

ERROR_REASON = 'error_reason'

ERROR_TYPE = 'error_type'

EVENT = 'event'

FUNCTION = 'function'

ISO_VERSION = 'iso_version'

LEVEL = 'level'

LOGGER_NAME = 'logger'

MESSAGE_TYPE = 'message_type'

MODULE = 'module'

RECORD_COUNT = 'record_count'

REQUEST_ID = 'request_id'

SCHEMA_PATH = 'schema_path'

SIZE_BYTES = 'size_bytes'

TEMPLATE_PATH = 'template_path'

TIMESTAMP = 'timestamp'

TRANSACTION_COUNT = 'transaction_count'

VALIDATION_TYPE = 'validation_type'

VERSION = 'version'

class pain001.logging_schema.schema.LogLevel

Bases: object

Standard log level names for structured logging.

CRITICAL = 'CRITICAL'

DEBUG = 'DEBUG'

ERROR = 'ERROR'

INFO = 'INFO'

WARNING = 'WARNING'

pain001.logging_schema.tracker module

Execution summary tracking with automatic event counting.

class pain001.logging_schema.tracker.ExecutionSummaryTracker(logger: Logger, dry_run: bool = False, message_type: str | None = None)

Bases: object

Track execution metrics for final summary report.

This class provides automatic log event counting and execution metrics tracking for generating comprehensive summary reports. Use as a context manager for automatic start/end tracking.

Parameters:

• logger – Logger instance to use for summary report.

• dry_run – Whether this is a dry-run execution.

• message_type – ISO 20022 message type (if applicable).

Example

>>> with ExecutionSummaryTracker(logger) as tracker:
...     # Your execution logic here
...     tracker.increment_processed_records(1250)
...     tracker.set_validation_result("schema_validation", "PASSED")
# Summary report automatically logged on exit

>>> # Or use manually:
>>> tracker = ExecutionSummaryTracker(logger, dry_run=True)
>>> tracker.start()
>>> # ... execution logic ...
>>> tracker.log_summary()

abort() → None

Mark execution as aborted.

increment_event_count(level: str) → None

Increment count for a specific log level.

Parameters:

level – Log level name (debug, info, warning, error, critical).

increment_processed_records(count: int = 1) → None

Increment total records processed count.

Parameters:

count – Number of records to add (default: 1).

log_summary() → None

Log execution summary report.

set_log_file(file_path: str) → None

Set log file path.

Parameters:

file_path – Path to log file.

set_output_file(file_path: str | None) → None

Set output file path.

Parameters:

file_path – Path to generated output file (None for dry-run).

set_validation_result(validation_type: str, result: str) → None

Set validation result for a specific validation type.

Parameters:

• validation_type – Type of validation (e.g., “schema_validation”).

• result – Result status (e.g., “PASSED”, “FAILED”).

start() → None

Mark execution start time.