pain001 package#

Subpackages#

Module contents#

The Python pain001 module.

class pain001.ConfigManager[source]#

Bases: object

Load and merge project, user, env, preset, and CLI configuration.

discover_project_config() → Path | None[source]#

Return the first project config found in the cwd.

discover_user_config() → Path | None[source]#

Return the default user config path if present.

get_profile(profile_name: str) → dict[str, Any][source]#

Return a built-in preset or user-defined profile.

load_from_file(path: str | Path) → dict[str, Any][source]#

Load one config file in YAML, TOML, or INI format.

resolve(cli_args: Mapping[str, Any]) → dict[str, Any][source]#

Merge defaults, user/project config, env, profile, and CLI args.

exception pain001.DataSourceError[source]#

Bases: Pain001Error

Raised when data source access fails.

This exception indicates issues with: - File not found (CSV, SQLite) - Database connection errors - Corrupted data files - Unsupported file formats - Empty data sources

Example

>>> try:
...     load_payment_data("payments.csv")
... except DataSourceError as e:
...     # Data access error - check file exists
...     log.error(f"Cannot access data source: {e}")
class pain001.MetricEvent(name: str, timestamp: float, attributes: dict[str, ~typing.Any]=<factory>, trace_context: dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Structured event emitted from processing steps.

attributes: dict[str, Any]#
name: str#
timestamp: float#
trace_context: dict[str, Any]#
exception pain001.PaymentValidationError(message: str, field: str | None = None)[source]#

Bases: Pain001Error

Raised when payment data validation fails.

This exception indicates issues with input data such as: - Invalid IBAN format - Invalid BIC/SWIFT code - Invalid amount (negative, too large, wrong format) - Missing required fields (debtor name, creditor account, etc.) - Invalid date formats

Parameters:

  • message – Human-readable error message.

  • field – Optional field name that caused the validation error.

Example

>>> try:
...     validate_payment_data(data)
... except PaymentValidationError as e:
...     # User-facing error - show validation message
...     return {"error": str(e), "field": e.field}
class pain001.SchemeValidationResult(profile: str, violations: list[SchemeViolation] = <factory>)[source]#

Bases: object

Outcome of validating payment rows against a scheme profile.

profile#

Name of the profile that produced this result.

Type:

str

violations#

All violations found, in row order.

Type:

list[pain001.validation.schemes.SchemeViolation]

property is_valid: bool#

Whether the rows are free of error-severity violations.

Returns:

True when no "error" violations were found (warnings are allowed), False otherwise.

profile: str#
violations: list[SchemeViolation]#
class pain001.SchemeViolation(rule: str, message: str, index: int, field: str | None = None, severity: str = 'error')[source]#

Bases: object

A single scheme-rule breach found in a payment row.

rule#

Stable identifier of the rule (e.g. "SEPA-CCY").

Type:

str

message#

Human-readable description of the breach.

Type:

str

index#

Zero-based index of the offending payment row.

Type:

int

field#

Name of the offending field, when applicable.

Type:

str | None

severity#

"error" (scheme would reject) or "warning".

Type:

str

as_dict() → dict[str, Any][source]#

Return a JSON-serialisable representation of the violation.

Returns:

A dict with the rule, message, index, field, severity, and remediation hint.

field: str | None = None#
index: int#
message: str#
property remediation: str#

Return the remediation hint for this violation’s rule.

Returns:

The remediation hint, or an empty string if none is defined.

rule: str#
severity: str = 'error'#
class pain001.TemplateRegistry(templates_dir: Path = PosixPath('/opt/hostedtoolcache/Python/3.12.13/x64/lib/python3.12/site-packages/pain001/templates'))[source]#

Bases: object

Discover and query supported template bundles.

get_template(message_type: str) → TemplateMetadata[source]#

Return metadata for a supported message type.

list_supported_versions() → list[str][source]#

Return supported message types in sorted order.

list_templates() → list[TemplateMetadata][source]#

Return every discovered template.

resolve_paths(message_type: str) → tuple[str, str][source]#

Return bundled template and schema paths for a message type.

search_by_category(category: str) → list[TemplateMetadata][source]#

Return templates whose category matches case-insensitively.

pain001.build_camt053_statement(statement_id: str, iban: str, currency: str, entries: list[dict[str, Any]], *, electronic_sequence_number: str = '1', version: str = 'camt.053.001.02') → str[source]#

Build a camt.053 bank-to-customer statement as an XML string.

Parameters:

  • statement_id – Statement identifier (Stmt/Id).

  • iban – Account IBAN (Stmt/Acct/Id/IBAN).

  • currency – Account currency (Stmt/Acct/Ccy, ISO 4217).

  • entries – One dict per booked entry. Recognised keys: amount (required), credit_debit_indicator (CRDT/DBIT, required), status (BOOK/PDNG/INFO, default BOOK), booking_date, value_date, entry_reference, remittance_information, and an optional per-entry currency (defaults to the account currency). A ValueError is raised for an entry missing amount or carrying an invalid credit/debit indicator or status code.

  • electronic_sequence_number – Statement sequence number.

  • version – camt.053 message version (drives the namespace).

Returns:

The statement as a UTF-8 XML string (with declaration).

pain001.build_pain002_report(message_id: str, original_message_id: str, group_status: str, payment_statuses: list[dict[str, Any]], *, original_message_name_id: str = 'pain.001.001.03', creation_datetime: str | None = None, version: str = 'pain.002.001.03') → str[source]#

Build a pain.002 payment status report as an XML string.

Parameters:

  • message_id – Identifier for this status report (GrpHdr/MsgId).

  • original_message_id – The id of the pain.001 message being reported on (OrgnlGrpInfAndSts/OrgnlMsgId).

  • group_status – Group-level status code (e.g. ACCP, RJCT).

  • payment_statuses – One dict per original payment information block. Recognised keys: original_payment_information_id (required), payment_information_status (required), and the optional transaction triplet original_end_to_end_id, transaction_status, status_reason.

  • original_message_name_id – Name id of the original message (default pain.001.001.03).

  • creation_datetime – ISO-8601 creation timestamp; defaults to now (UTC).

  • version – pain.002 message version (drives the namespace).

Returns:

The pain.002 report as a UTF-8 XML string (with declaration).

Raises:

ValueError – If group_status or any row status is not a known ISO 20022 status code, or a required row key is missing.

pain001.canonicalize_payment_record(row: dict[str, Any]) → dict[str, Any][source]#

Map a record’s alias keys to canonical field names, preserving values.

The key-mapping half of normalize_payment_records(): amount becomes payment_amount, currency and payment_currency mirror each other, and lower-case *_iban / *_bic spellings are canonicalized - without reformatting any value. Use this before JSON-Schema validation, where typed values (numbers, booleans) must keep their types.

Parameters:

row – The raw input record.

Returns:

A new dict with canonical keys; the input is not mutated.

pain001.clear_metrics_callbacks() → None[source]#

Remove all registered metric callbacks.

pain001.generate_xml_string(data: list[dict[str, object]], payment_initiation_message_type: str, xml_template_path: str, xsd_schema_path: str) → str[source]#

Generate ISO 20022 pain.001 XML content as a string (in-memory).

This function is ideal for serverless architectures, REST APIs, and microservices where XML needs to be returned without writing to disk.

Parameters:

  • data – List of dictionaries containing payment data.

  • payment_initiation_message_type – Message type (e.g., “pain.001.001.03”).

  • xml_template_path – Path to the Jinja2 XML template file.

  • xsd_schema_path – Path to XSD schema file for validation.

Returns:

The generated and validated XML content.

Return type:

str

Raises:

  • ValueError – If message type is invalid, data is empty, the template or schema path fails validation, or the template contains disabled Jinja filesystem directives.

  • RuntimeError – If XML validation fails against XSD schema.

PaymentValidationError from amount normalization (missing, non-numeric, non-positive, or over-precise payment_amount values) propagates unchanged.

Examples

>>> data = [{"id": "MSG001", "date": "2026-01-15", ...}]
>>> xml_str = generate_xml_string(
...     data,
...     "pain.001.001.03",
...     "templates/pain.001.001.03/template.xml",
...     "templates/pain.001.001.03/pain.001.001.03.xsd"
... )
>>> xml_str.startswith('<?xml')
True
async pain001.generate_xml_string_async(data: list[dict[str, object]], payment_initiation_message_type: str, xml_template_path: str, xsd_schema_path: str) → str[source]#

Run generate_xml_string() in a worker thread.

pain001.main(xml_message_type: str | None, xml_template_file_path: str | None, xsd_schema_file_path: str | None, data_file_path: str | None, dry_run: bool = False) → None[source]#

Main entry point for python -m pain001.

Parameters:

  • xml_message_type – ISO 20022 message type (e.g., ‘pain.001.001.03’).

  • xml_template_file_path – Path to Jinja2 XML template file.

  • xsd_schema_file_path – Path to XSD schema for validation.

  • data_file_path – Path to CSV or SQLite data file.

  • dry_run – If True, validate inputs without generating XML.

Exits:

0 on success, 1 on validation or processing error.

pain001.normalize_payment_records(data: list[dict[str, Any]]) → list[dict[str, Any]][source]#

Normalize caller records into the canonical pain.001 input shape.

Applies the same ergonomics the generators use internally, so records can also be pre-normalized before JSON-Schema validation:

  • field aliases (amount -> payment_amount, currency <-> payment_currency, lower-case IBAN/BIC key spellings, …);

  • amounts formatted as exact two-decimal strings;

  • Python/JSON booleans (and "True"/"FALSE" strings) rendered in XSD form ("true"/"false");

  • bare dates coerced to the XSD lexical form each field requires (date -> YYYY-MM-DDT00:00:00, requested_execution_date truncated to YYYY-MM-DD);

  • nb_of_txs and ctrl_sum computed from the rows themselves.

Parameters:

data – The raw payment rows.

Returns:

A new list of normalized rows; the input is not mutated.

A PaymentValidationError from amount normalization (missing, non-numeric, non-positive, or over-precise payment amounts) propagates unchanged.

pain001.parse_camt053_statement(xml_file_path: str, xsd_file_path: str | None = None) → dict[str, object][source]#

Parse a camt.053 statement into a compact Python structure.

pain001.parse_pain002_report(xml_file_path: str, xsd_file_path: str | None = None) → dict[str, object][source]#

Parse a pain.002 payment status report into structured data.

pain001.process_files(xml_message_type: str, xml_template_file_path: str, xsd_schema_file_path: str, data_file_path: str | list[dict[str, Any]] | dict[str, Any], output_path: str | None = None) → str[source]#

Generate an ISO 20022 payment message from various data sources.

Parameters:

  • xml_message_type – XML message type (e.g., ‘pain.001.001.03’).

  • xml_template_file_path – Path to the XML template file.

  • xsd_schema_file_path – Path to the XSD schema file.

  • data_file_path – File path (CSV/DB/JSON/Parquet) or Python data (list/dict).

  • output_path – Explicit path for the generated XML file. When omitted, the file is written next to the template (deprecated; requires the template to live under the current working directory).

Returns:

The path the generated XML file was written to.

Raises:

  • XMLGenerationError – If the message type is not supported or the XML file could not be written.

  • Exception – Any error raised while validating inputs, loading data, or generating XML (e.g. FileNotFoundError for missing files, PaymentValidationError for bad amounts) is logged and re-raised unchanged.

async pain001.process_files_async(xml_message_type: str, xml_template_file_path: str, xsd_schema_file_path: str, data_file_path: str) → str[source]#

Run process_files() in a worker thread.

async pain001.process_files_streaming_async(xml_message_type: str, xml_template_file_path: str, xsd_schema_file_path: str, data_file_path: str, chunk_size: int = 1000) → list[str][source]#

Run process_files_streaming() in a worker thread.

pain001.register_metrics_callback(fn: Callable[[MetricEvent], None]) → None[source]#

Register a callback for structured metric events.

pain001.sanitize_to_charset(value: str, replacement: str = ' ') → str[source]#

Transliterate value into the ISO 20022 character set.

Accented Latin letters are decomposed to their base letter (é becomes e, ü becomes u). Any character that still falls outside the permitted set is replaced with replacement.

Parameters:

  • value – The text to transliterate.

  • replacement – The string substituted for characters that cannot be transliterated (default: a single space).

Returns:

A string containing only ISO 20022 permitted characters.

Example

>>> sanitize_to_charset("Café Münchën")
'Cafe Munchen'
>>> sanitize_to_charset("A & B", replacement="and")
'A and B'
async pain001.validate_all_async(validation_service: ValidationService, config: ValidationConfig) → Any[source]#

Run ValidationService.validate_all() in a worker thread.

pain001.validate_registry(registry: TemplateRegistry = <pain001.templates.registry.TemplateRegistry object>) → list[str][source]#

Validate every registered template and return validated IDs.

pain001.validate_scheme(data: list[dict[str, Any]], profile: str = 'sepa-sct') → SchemeValidationResult[source]#

Validate payment rows against a named scheme profile.

Parameters:

  • data – Loaded payment rows (the normalised internal form).

  • profile – Profile name to apply (default: "sepa-sct").

Returns:

A SchemeValidationResult listing every violation.

Raises:

ValueError – If profile is not a registered profile name.

Example

>>> rows = [{
...     "payment_currency": "USD",
...     "debtor_account_IBAN": "DE89370400440532013000",
...     "creditor_account_IBAN": "FR1420041010050500013M02606",
...     "payment_amount": "100.00",
... }]
>>> result = validate_scheme(rows, profile="sepa-sct")
>>> result.is_valid
False
>>> result.violations[0].rule
'SEPA-CCY'