API Reference#

This page documents the public Python API of camt053. Every developer surface (CLI, REST API, MCP, and LSP) is built on the shared camt053.services facade, so the functions documented here behave identically across interfaces.

Top-level package#

The Python camt053 module.

camt053.parse_document(xml)[source]#

Parse a camt.05x statement document into the typed model.

Parameters:: xml (str) – The raw statement XML as a string.
Return type:: ParsedDocument
Returns:: A ParsedDocument.
Raises:: StatementParseError – If the XML is malformed or its root is not a recognised camt.052 / camt.053 / camt.054 document.

camt053.parse_statement(xml)[source]#

Parse a document and return its first statement.

Convenience for the common single-statement case.

Parameters:: xml (str) – The raw statement XML as a string.
Return type:: Statement
Returns:: The first Statement in the document.
Raises:: StatementParseError – If the XML cannot be parsed or contains no statement element.

camt053.serialize_document(document)[source]#

Re-serialise a parsed document back to validated camt.053 XML.

Parameters:: document (ParsedDocument) – A parsed ParsedDocument.
Return type:: str
Returns:: The validated camt.053.001.14 statement document as a string.
Raises:: XMLGenerationError – If the document carries no statement, or the rendered XML does not validate against the bundled XSD.

camt053.serialize_statement(statement)[source]#

Re-serialise a single parsed statement to validated camt.053 XML.

Convenience wrapper that wraps statement in a one-statement document.

Parameters:: statement (Statement) – A parsed Statement.
Return type:: str
Returns:: The validated camt.053.001.14 statement document as a string.
Raises:: XMLGenerationError – If the rendered XML does not validate.

camt053.generate_reversal_xml(records, output_format='camt053', version=None)[source]#

Render reversing-entry records into a validated reversal document.

Parameters:

records (list[dict[str, Any]]) – One or more flat reversing-entry records (as produced by camt053.reversal.reversal.build_reversal_records()).
output_format (str) – "camt053" (default) emits a camt.053 reversing-entry statement; "pacs004" emits a pacs.004 PaymentReturn document.
version (str | None) – For the camt.053 format, the schema version to emit (one of camt053.constants.REVERSAL_CAMT_VERSIONS); None keeps the default camt053.constants.REVERSAL_MESSAGE_TYPE. Ignored for the pacs.004 format.

Return type:

str

Returns:

The validated reversal document as a string.

Raises:

ReversalGenerationError – If records are empty, the format / version is unknown, or the rendered XML does not validate against its XSD.

camt053.generate_reversal_for_statement(statement, reason_code=None, msg_id=None, creation_date_time=None, output_format='camt053', version=None)[source]#

Build and render a reversing-entry statement for a parsed statement.

Convenience that chains camt053.reversal.reversal.build_reversal_records() and generate_reversal_xml().

Parameters:

statement (Statement) – The parsed statement whose entries are reversed.
reason_code (str | None) – If given, only entries with this return reason are reversed; otherwise every returnable entry is.
msg_id (str | None) – Optional reversal group-header message id.
creation_date_time (str | None) – Optional ISO 8601 timestamp for the reversal.
output_format (str) – "camt053" (default) or "pacs004".
version (str | None) – Optional camt.053 schema version to emit (default keeps camt053.constants.REVERSAL_MESSAGE_TYPE).

Return type:

str

Returns:

The validated reversal document as a string.

Raises:

ReversalGenerationError – If no entry matches or validation fails.

class camt053.Entry(reference=None, amount=None, currency=None, credit_debit_indicator=None, status=None, booking_date=None, value_date=None, account_servicer_ref=None, reversal_indicator=False, reason_code=None, details=<factory>)[source]#

Bases: object

A booked entry (Ntry) on a statement.

reason_code is a convenience: it surfaces the first return reason found among the entry’s transaction details, so callers can filter entries without descending into details.

Parameters:

reference (str | None)
amount (str | None)
currency (str | None)
credit_debit_indicator (str | None)
status (str | None)
booking_date (str | None)
value_date (str | None)
account_servicer_ref (str | None)
reversal_indicator (bool)
reason_code (str | None)
details (list[TransactionDetails])

reference: str | None = None#

amount: str | None = None#

currency: str | None = None#

credit_debit_indicator: str | None = None#

status: str | None = None#

booking_date: str | None = None#

value_date: str | None = None#

account_servicer_ref: str | None = None#

reversal_indicator: bool = False#

reason_code: str | None = None#

details: list[TransactionDetails]#

property amount_decimal: Decimal | None#

Return amount as a Decimal, or None if invalid.

The string amount is preserved for XML fidelity; this is a convenience for arithmetic. Empty or unparsable values yield None.

is_returnable()[source]#

Return True if this entry carries any return reason code.

Return type:: bool

to_dict()[source]#

Return a JSON-serialisable representation.

Return type:: dict[str, Any]

class camt053.Statement(id=None, electronic_seq_nb=None, creation_date_time=None, account=<factory>, balances=<factory>, entries=<factory>)[source]#

Bases: object

A single statement / report / notification within a document.

Parameters:

id (str | None)
electronic_seq_nb (str | None)
creation_date_time (str | None)
account (Account)
balances (list[Balance])
entries (list[Entry])

id: str | None = None#

electronic_seq_nb: str | None = None#

creation_date_time: str | None = None#

account: Account#

balances: list[Balance]#

entries: list[Entry]#

entries_with_reason(reason_code)[source]#

Return entries whose return reason matches reason_code.

The match is case-insensitive and considers both the entry-level convenience reason and every transaction detail.

Parameters:: reason_code (str) – An ISO external return reason code (e.g. "AC04").
Return type:: list[Entry]

to_dict()[source]#

Return a JSON-serialisable representation.

Return type:: dict[str, Any]

class camt053.ParsedDocument(message_type, msg_id=None, creation_date_time=None, statements=<factory>)[source]#

Bases: object

A parsed camt.05x document: its group header and statements.

Parameters:

message_type (str)
msg_id (str | None)
creation_date_time (str | None)
statements (list[Statement])

message_type: str#

msg_id: str | None = None#

creation_date_time: str | None = None#

statements: list[Statement]#

all_entries()[source]#

Return every entry across all statements in the document.

Return type:: list[Entry]

to_dict()[source]#

Return a JSON-serialisable representation.

Return type:: dict[str, Any]

exception camt053.StatementParseError(message, line=None)[source]#

Bases: Camt053Error

Raised when an incoming camt.05x statement cannot be parsed.

This exception indicates issues with: - Malformed or non-well-formed statement XML - A document whose root is not a recognised camt.05x message - Missing mandatory statement structure (group header, statement, entries)

Example

>>> try:
...     parse_document(xml)
... except StatementParseError as e:
...     # Bad inbound file - quarantine and alert operations
...     log.error(f"Cannot parse statement: {e}")

Parameters:

message (str)
line (int | None)

code: str = 'STATEMENT_PARSE_ERROR'#

exception camt053.ReversalGenerationError[source]#

Bases: XMLGenerationError

Raised when a reversing entry cannot be generated.

This exception indicates issues with: - No statement entries matched the requested return/reason code - A booked entry missing the fields required to reverse it - Rendering or XSD validation failure of the reversal document

Example

>>> try:
...     generate_reversal(xml, reason_code="AC04")
... except ReversalGenerationError as e:
...     # Nothing to reverse, or reversal failed validation
...     log.error(f"Reversal generation failed: {e}")

code: str = 'REVERSAL_GENERATION_ERROR'#

Services facade#

High-level service facade for Camt053.

A small, dependency-light layer that exposes the library’s core capabilities as plain functions returning plain data. It is the single shared backend for the CLI, the REST API, the Model Context Protocol (MCP) server, and the Language Server Protocol (LSP) server, so every interface behaves identically.

The headline capability is the one-shot reversing-entry workflow: read an incoming camt.053 statement, find the entries carrying a return reason code (e.g. AC04 Closed Account), and emit a validated reversing entry.

Example

>>> from camt053 import services
>>> [m["message_type"] for m in services.list_message_types()][:1]
['camt.052.001.08']
>>> services.validate_identifier("bic", "NWBKGB2LXXX")["valid"]
True

camt053.services.list_message_types()[source]#

Return every supported message type with its human-readable name.

Return type:: list[dict[str, str]]
Returns:: A list of {"message_type": ..., "name": ...} dictionaries.

camt053.services.list_return_reasons()[source]#

Return every known ISO external return reason code with its name.

Return type:: list[dict[str, str]]
Returns:: A list of {"code": ..., "name": ...} dictionaries.

camt053.services.validate_reason_code(code)[source]#

Validate an ISO external return reason code.

Parameters:: code (str) – An ISO external return reason code (case-insensitive).
Return type:: dict[str, Any]
Returns:: {"code": str, "name": str, "valid": bool}. Unknown codes report valid=False with the generic "Unknown reason code" name.

camt053.services.classify_reason(code, overrides=None, default=None)[source]#

Classify a return reason code into a handling action.

Maps an ISO external return reason code to "return", "retry", or "ignore" using the built-in policy (account-level rejections return, transient conditions retry, informational reasons ignore). The lookup is case-insensitive; unknown / unmapped codes resolve to default.

Parameters:

code (str) – An ISO external return reason code (case-insensitive).
overrides (dict[str, str] | None) – Optional code -> action mapping taking precedence over the built-in policy (keys matched case-insensitively).
default (str | None) – The action for codes the policy / overrides do not cover (defaults to "return").

Return type:

dict[str, str]

Returns:

{"code": str, "name": str, "action": str}.

camt053.services.reason_policy(overrides=None, default=None)[source]#

Return the full reason-code action policy.

Parameters:

overrides (dict[str, str] | None) – Optional code -> action mapping taking precedence over the built-in policy (keys matched case-insensitively).
default (str | None) – The action for codes the policy / overrides do not cover (defaults to "return").

Return type:

dict[str, Any]

Returns:

{"default": str, "actions": [str, ...], "policy": {code: action}}.

camt053.services.get_input_schema(message_type)[source]#

Return the JSON Schema for a message type’s reversing-entry record.

Parameters:: message_type (str) – A supported ISO 20022 cash-management message type.
Return type:: dict[str, Any]
Returns:: The parsed JSON Schema document.
Raises:: ValueError – If the message type is not supported.

camt053.services.get_required_fields(message_type)[source]#

Return the required reversing-entry field names for a message type.

Parameters:: message_type (str) – A supported ISO 20022 cash-management message type.
Return type:: list[str]
Returns:: The list of required field names.
Raises:: ValueError – If the message type is not supported.

camt053.services.validate_records(message_type, records)[source]#

Validate flat reversing-entry records against a type’s input schema.

Parameters:

message_type (str) – A supported ISO 20022 cash-management message type.
records (list[dict[str, Any]]) – One or more flat reversing-entry records.

Return type:

dict[str, Any]

Returns:

A report dict with keys valid, total, valid_count, and errors – a list of dicts with keys row, path, and message.

Raises:

ValueError – If the message type is not supported.

camt053.services.validate_identifier(kind, value)[source]#

Validate a financial identifier (IBAN, BIC, or LEI).

Parameters:

kind (str) – One of "iban", "bic", or "lei" (case-insensitive).
value (str) – The identifier value to check.

Return type:

dict[str, Any]

Returns:

{"kind": str, "value": str, "valid": bool}.

Raises:

ValueError – If kind is not a supported identifier type.

camt053.services.validate_currency(code)[source]#

Validate an ISO 4217 alphabetic currency code.

The check is case-insensitive. The returned minor_units is the number of decimal places in the currency’s subdivision (e.g. EUR=2, JPY=0), or None when the code is unknown.

Parameters:: code (str) – A three-letter ISO 4217 currency code.
Return type:: dict[str, Any]
Returns:: {"code": str, "valid": bool, "minor_units": int | None}.

camt053.services.parse_statement(xml)[source]#

Parse an incoming camt.05x statement into plain data.

Parameters:: xml (str) – The raw statement XML as a string.
Return type:: dict[str, Any]
Returns:: The parsed document as a JSON-serialisable dict (group header plus statements, each with its account, balances, and entries).
Raises:: StatementParseError – If the XML is malformed or unrecognised.

camt053.services.serialize_statement(xml)[source]#

Round-trip a statement: parse it, then re-serialise it to camt.053 XML.

Parses the incoming camt.05x statement into the typed model and renders it back to a validated camt.053.001.14 document. The result is deterministic and preserves the account, balances, and entries (their references, amounts, currencies, credit/debit indicators, and return reasons), so parse_statement(serialize_statement(xml)) reproduces the same parsed data.

Parameters:

xml (str) – The raw statement XML as a string.

Return type:

str

Returns:

The re-serialised, XSD-validated camt.053.001.14 document as a string.

Raises:

StatementParseError – If the XML is malformed or unrecognised.
XMLGenerationError – If the document carries no statement or the rendered XML does not validate against the bundled XSD.

camt053.services.validate_statement(xml)[source]#

Validate an incoming statement against its official ISO camt XSD.

Detects the document’s message type from its namespace and validates it against the matching official XSD bundled with the package.

Parameters:: xml (str) – The raw statement XML as a string.
Return type:: dict[str, Any]
Returns:: {"valid": bool, "message_type": str, "errors": [str, ...]}.
Raises:: StatementParseError – If the XML is malformed / unrecognised, or no official XSD is bundled for the detected message type.

camt053.services.list_entries(xml, *, streaming=False)[source]#

Parse a statement and return every entry across all its statements.

With streaming=False (default) the whole tree is parsed at once. With streaming=True the entries are read incrementally via camt053.parse.statement_parser.iter_statement_entries(), which bounds peak memory to a single entry; the returned list is identical either way. Callers that want to consume entries one at a time without materialising the full list should use iter_entries() instead.

Parameters:

xml (str) – The raw statement XML as a string.
streaming (bool) – If True, use the memory-bounded iterparse path.

Return type:

list[dict[str, Any]]

Returns:

A list of entry dicts.

Raises:

StatementParseError – If the XML is malformed or unrecognised.

camt053.services.iter_entries(xml)[source]#

Stream a statement’s entries as dicts without building the whole tree.

Memory-bounded, forward-only counterpart to list_entries(): each entry dict is produced the moment its <Ntry> element closes (via camt053.parse.statement_parser.iter_statement_entries()), so large statements can be processed without loading the entire document. For a well-formed statement it yields exactly the entries list_entries() returns, in document order.

Parameters:: xml (str) – The raw statement XML as a string.
Yields:: Each entry as a JSON-serialisable dict, in document order.
Raises:: StatementParseError – If the XML is empty or becomes malformed while streaming (after any already-parsed entries have been yielded).
Return type:: Iterator[dict[str, Any]]

camt053.services.filter_entries(xml, reason_code='AC04', *, status=None, date_from=None, date_to=None, min_amount=None, max_amount=None)[source]#

Return statement entries matching every supplied filter (ANDed).

Called as before, filter_entries(xml) keeps the default-AC04 reason behaviour. Passing reason_code=None (or an empty string) drops the reason filter, so the other criteria can be used on their own.

Parameters:

xml (str) – The raw statement XML as a string.
reason_code (str | None) – ISO external return reason to match (default "AC04"); None / empty disables the reason filter.
status (str | None) – If given, only entries with this status (e.g. "BOOK").
date_from (str | None) – If given, only entries booked on or after this ISO date.
date_to (str | None) – If given, only entries booked on or before this ISO date.
min_amount (str | None) – If given, only entries with an amount >= this value.
max_amount (str | None) – If given, only entries with an amount <= this value.

Return type:

list[dict[str, Any]]

Returns:

A list of matching entry dicts.

Raises:

StatementParseError – If the XML is malformed or unrecognised.
ValueError – If a date or amount filter value is invalid.

camt053.services.build_reversal(xml, reason_code='AC04')[source]#

Build the flat reversing-entry records for a statement’s matches.

Parses the statement, selects every statement’s entries with the given return reason, and maps them to flat records (without rendering XML).

Parameters:

xml (str) – The raw statement XML as a string.
reason_code (str) – The ISO external return reason to reverse (default "AC04").

Return type:

list[dict[str, Any]]

Returns:

A list of flat reversing-entry records.

Raises:

StatementParseError – If the XML cannot be parsed.
ReversalGenerationError – If no entry matches the reason code.

camt053.services.cleanse_records(records)[source]#

Cleanse the SWIFT-constrained fields of reversing-entry records.

Transliterates or strips characters outside the SWIFT X charset and enforces the maximum length of every name / narrative field (Nm / AddtlInf / party / counterparty names), mutating the records in place so they are safe to render onto SWIFT FIN / CBPR+ rails.

Parameters:: records (list[dict[str, Any]]) – The flat reversing-entry records to cleanse in place.
Return type:: dict[str, Any]
Returns:: {"changed": int, "fields": [report, ...]} where each report is a to_dict() of a field whose value was altered (unchanged fields are omitted).

camt053.services.generate_reversal(xml, reason_code='AC04', msg_id=None, creation_date_time=None, cleanse=False, output_format='camt053', version=None)[source]#

Read a statement and generate a validated reversing-entry document.

This is the headline one-shot workflow: parse the incoming camt.053, pick the entries with the requested return reason (e.g. AC04 Closed Account), and emit a validated reversal document.

By default the reversal is a camt.053.001.14 reversing-entry statement. Callers can select a different bundled camt.053 schema version via version (one of camt053.constants.REVERSAL_CAMT_VERSIONS), or switch to a pacs.004 PaymentReturn document with output_format="pacs004".

Parameters:

xml (str) – The raw incoming statement XML as a string.
reason_code (str) – The ISO external return reason to reverse (default "AC04").
msg_id (str | None) – Optional reversal group-header message id.
creation_date_time (str | None) – Optional ISO 8601 timestamp for the reversal.
cleanse (bool) – Opt-in SWIFT charset cleansing of the reversal’s name / narrative fields before rendering (default False, leaving the existing output byte-for-byte unchanged).
output_format (str) – "camt053" (default) emits a camt.053 reversing-entry statement; "pacs004" emits a pacs.004 PaymentReturn document.
version (str | None) – For the camt.053 format, the schema version to emit; None keeps the default camt053.constants.REVERSAL_MESSAGE_TYPE.

Return type:

str

Returns:

The validated reversal document as a string.

Raises:

StatementParseError – If the XML cannot be parsed.
ReversalGenerationError – If no entry matches, the format / version is unknown, or validation fails.

camt053.services.generate(records, cleanse=False, output_format='camt053', version=None)[source]#

Render flat reversing-entry records into a validated reversal document.

The in-process entry point for callers that already hold reversing-entry records (e.g. built elsewhere or loaded from a data file).

Parameters:

records (list[dict[str, Any]]) – One or more flat reversing-entry records.
cleanse (bool) – Opt-in SWIFT charset cleansing of the records’ name / narrative fields before rendering (default False).
output_format (str) – "camt053" (default) or "pacs004".
version (str | None) – Optional camt.053 schema version to emit (default keeps camt053.constants.REVERSAL_MESSAGE_TYPE).

Return type:

str

Returns:

The validated reversal document as a string.

Raises:

ReversalGenerationError – If records are empty, the format / version is unknown, or validation fails.

camt053.services.generate_batch(paths, reason_code='AC04', cleanse=False, output_format='camt053', version=None)[source]#

Generate reversals for many statement files in one call.

Processes a list of file paths, a glob pattern, or a directory (every *.xml file inside it, recursively), reversing each statement independently. Each file is isolated: a parse / generation failure on one file is captured as an error result and does not abort the batch.

Parameters:

paths (list[str] | str) – A directory path, a glob pattern, a single file path, or a list mixing any of those. Directories are scanned recursively for *.xml files; bare paths that are neither a file, directory, nor matching glob are reported as a not-found error.
reason_code (str) – The ISO external return reason to reverse (default "AC04").
cleanse (bool) – Opt-in SWIFT charset cleansing before rendering.
output_format (str) – "camt053" (default) or "pacs004".
version (str | None) – Optional camt.053 schema version to emit.

Return type:

dict[str, Any]

Returns:

A summary dict with keys total, succeeded, failed, and results – a list of per-file dicts with keys path, ok, xml, and error. Results preserve discovery order.

camt053.services.configure_logging(*, level=20, json_format=True, stream=None)[source]#

Opt in to Camt053 logging on the library’s named logger.

Attaches a single logging.StreamHandler to the "camt053" logger, replacing any handler a previous call installed (so repeated calls are idempotent and never duplicate output). Propagation to the root logger is disabled so the library’s records do not leak into an unconfigured root.

Parameters:

level (int) – The logging level to set on the logger (default INFO).
json_format (bool) – When True (default) emit structured JSON via JsonFormatter; when False emit a human-readable line.
stream (Any | None) – The stream to write to (defaults to sys.stderr via logging.StreamHandler).

Return type:

Logger

Returns:

The configured "camt053" logger.

camt053.services.configure_logging_from_env()[source]#

Configure logging from the CAMT053_LOG_FORMAT environment variable.

Reads CAMT053_LOG_FORMAT ("json" or "text", case-insensitive) and, when set to a recognised value, calls configure_logging() accordingly. The level is taken from CAMT053_LOG_LEVEL (a standard level name, default INFO). When the variable is unset or empty, logging is left untouched and None is returned, preserving the off-by-default behaviour.

Return type:: Logger | None
Returns:: The configured logger, or None when no recognised format was requested.

camt053.services.guard_xml(xml, max_bytes=1048576)[source]#

Reject an untrusted XML payload that breaches a security limit.

A thin facade over camt053.security.xml_guard.guard_xml_payload(), enforcing a maximum byte size and refusing inline DTD / entity declarations before the payload reaches the parser.

Parameters:

xml (str) – The raw XML payload as a string.
max_bytes (int) – The maximum accepted UTF-8 size in bytes.

Raises:

XmlSecurityError – If the payload is too large or carries a DOCTYPE / ENTITY declaration.

Return type:

None

camt053.services.check_cbpr_readiness(xml)[source]#

Check a camt.053 payload for CBPR+ Nov 2026 readiness.

The checker walks the document and reports:

Schema version vs the CBPR+ current set (camt.053.001.08 / camt.053.001.13). Deprecated versions are flagged as warnings, unknown versions as errors.
Postal addresses: every <PstlAdr> element is classified as fully structured (no <AdrLine>), hybrid (both <AdrLine> and <Twn>/<Ctry>), or unstructured-only (one or more <AdrLine> with no <Twn>/<Ctry> sibling). Unstructured-only is the Nov 2026 reject case.

Parameters:

xml (str) – The camt.053 payload as a string. The same byte-cap and DOCTYPE/ENTITY pre-flight enforced by camt053.security.xml_guard.guard_xml_payload() apply here.

Returns:

{
    "cbpr_ready": bool,
    "schema_version": str | None,
    "checked_at": "<ISO 8601 UTC timestamp>",
    "cutover_date": "2026-11-16",
    "issues": [
        {
            "code": "<STABLE_CODE>",
            "severity": "error" | "warning",
            "path": "<XPath-style location>",
            "message": "<human-readable explanation>",
        },
        ...
    ],
    "summary": {
        "addresses_checked": int,
        "fully_structured": int,
        "hybrid": int,
        "unstructured_only": int,
    },
}

cbpr_ready is True iff no severity="error" issue was raised.

Return type:

A dictionary with the shape

Raises:

camt053.security.xml_guard.XmlSecurityError – If the payload exceeds the byte cap or carries a DOCTYPE / ENTITY declaration.
ValueError – If the payload is not well-formed XML.

camt053.services.compute_dedupe_key(xml)[source]#

Return the canonical dedupe key for an incoming statement’s first stmt.

Convenience wrapper that parses the payload and returns the "{MsgId}:{StmtId}:{ElctrncSeqNb}" key. Two payloads that share this key represent the same statement (typically a bank replay) and should be processed at most once downstream.

Parameters:

xml (str) – The raw camt.05x statement XML as a string.

Return type:

str

Returns:

The colon-joined dedupe key for the first statement.

Raises:

ValueError – If the document carries zero statements.
StatementParseError – If the XML is malformed or unrecognised.

camt053.services.compute_dedupe_keys(xml)[source]#

Return one dedupe key per statement bundled in the document.

Useful for multi-statement documents (e.g. a daily multi-account report). The first key matches compute_dedupe_key()’s output.

Parameters:: xml (str) – The raw camt.05x statement XML as a string.
Return type:: list[str]
Returns:: A list of dedupe keys in document order; empty when no statements are present.
Raises:: StatementParseError – If the XML is malformed or unrecognised.

camt053.services.stable_reversal_reference(original_reference, *, max_length=35)[source]#

Return a stable, idempotent reversal reference for an original entry.

Deterministic: the same input always produces byte-identical output. The default scheme is a human-readable RVSL-{original} prefix (preserved across releases for navigability in audit logs); when the prefixed form would exceed max_length it falls back to a RVSL-{sha256(original|REV)} truncated digest, ensuring collision-resistant IDs even for very long or adversarial originals.

Either way the contract is: same input → identical output, bit for bit. Pinned by the property tests; do not change the prefix or the salt without bumping the major version.

Parameters:

original_reference (str) – The original booked entry’s reference (e.g. NTRY-0001). An empty string is permitted and yields the bare "RVSL-" prefix.
max_length (int) – ISO 20022 ID maximum length. Defaults to 35.

Return type:

str

Returns:

The reversal reference, never longer than max_length.

camt053.services.parse_statement_lenient(xml)[source]#

Parse an incoming statement, skipping corrupt entries instead of failing.

Strict mode (parse_statement()) aborts on the first per-entry parse failure — the safe default. Lenient mode is for batch jobs that prefer to process the surviving entries and surface the skipped ones, rather than abandon a 10,000-entry file because one <Ntry> is malformed.

Parameters:: xml (str) – The raw camt.05x statement XML as a string.
Return type:: dict[str, Any]
Returns:: A JSON-serialisable dict with keys document, corrupt_entry_count, and diagnostics – a list of dicts with keys stmt_index, entry_index, code, and message. document carries the same shape as parse_statement()’s return value; diagnostics is empty when no entries were skipped.
Raises:: StatementParseError – For document-level failures (empty XML, malformed XML, unrecognised root / container). These are not recoverable mid-stream.

camt053.services.compute_event_hmac(secret, *, prev_hash, sequence, timestamp_utc, event_type, payload)[source]#

Compute the HMAC-SHA-256 for an audit event under secret.

Exposed publicly so consumers can re-derive an event’s expected HMAC offline (e.g. during external verification) without having to instantiate a HashChain.

Parameters:

secret (bytes) – The chain’s HMAC secret. Must match what the producer used; mismatches yield HMAC_MISMATCH on verification.
prev_hash (str) – The previous event’s hmac, or GENESIS_HASH for the first event.
sequence (int) – Zero-based index of this event in the chain.
timestamp_utc (str) – ISO 8601 UTC with trailing Z.
event_type (str) – Application-defined event label.
payload (dict[str, Any]) – JSON-serialisable application payload.

Return type:

str

Returns:

The HMAC-SHA-256 as a 64-character lowercase hex string.

camt053.services.verify_chain(events, secret, *, starting_prev_hash='')[source]#

Verify that an AuditEvent list is an intact hash-chain.

Walks the events in order and checks two invariants per event:

HMAC integrity: the event’s stored hmac matches the HMAC re-derived from its other fields under secret.
Chain linkage: the event’s prev_hash equals the previous event’s hmac (or starting_prev_hash for the first event in the list).

The sequence indices are also checked: events must be consecutively numbered starting from zero (or from len(events) - len(events) if the caller is verifying a sub-segment, which is supported via starting_prev_hash).

Parameters:

events (list[AuditEvent]) – The chain to verify, in order.
secret (bytes) – The HMAC secret used by the producer.
starting_prev_hash (str) – The prev_hash the first event must carry. Defaults to GENESIS_HASH; pass the last hmac of an earlier segment to verify a continuation.

Return type:

ChainVerification

Returns:

A ChainVerification carrying the verdict.

camt053.services.telemetry = <module 'camt053.telemetry' from '/home/runner/work/camt053/camt053/camt053/telemetry.py'>#: Re-export the telemetry module for services.telemetry access. Already imported above; __all__ carries "telemetry" so the import-star surface includes it.

camt053.services.classify_schema_version(version)[source]#

Bucket a detected version into one of the four stable codes.

Parameters:: version (str | None) – The version string returned by detect_schema_version(), or None.
Return type:: str
Returns:: One of the SchemaClassification constants ("current" / "deprecated" / "unknown" / "unsupported"). None input always maps to SchemaClassification.UNSUPPORTED.

camt053.services.detect_schema_version(xml)[source]#

Return the camt.05x schema version from a payload’s root xmlns.

Walks just the root element (cheap, no full parse) and pulls the version string out of the namespace URI. Hostile / malformed input is rejected by the existing camt053.security.xml_guard.guard_xml_payload() and defusedxml pre-flight, so this helper is safe to call on untrusted bank files.

Parameters:

xml (str) – The raw XML payload as a string.

Return type:

str | None

Returns:

The detected version string (e.g. "camt.053.001.08"), or None if the root has no recognised camt.05x namespace.

Raises:

camt053.security.xml_guard.XmlSecurityError – For oversized payloads or inline DOCTYPE / ENTITY declarations.
ValueError – For payloads that are not well-formed XML.

camt053.services.validate_schema_version(xml, *, strict=False)[source]#

Pre-flight a camt.05x payload’s schema version.

Combines detect_schema_version() and classify_schema_version() into a single call returning a structured report. strict mode raises UnsupportedSchemaError on any classification other than SchemaClassification.CURRENT or SchemaClassification.DEPRECATED (i.e. unknown camt.05x minor revisions and non-camt.05x namespaces are refused).

Parameters:

xml (str) – The raw XML payload as a string.
strict (bool) – If True, refuse anything outside the current and deprecated sets. Defaults to False (report-only).

Return type:

dict[str, Any]

Returns:

A dictionary with the shape:

{
    "version": "camt.053.001.08" | None,
    "classification": "current" | "deprecated" |
                      "unknown" | "unsupported",
    "supported": bool,
}

supported is True iff the classification is SchemaClassification.CURRENT or SchemaClassification.DEPRECATED.

Raises:

UnsupportedSchemaError – In strict mode, when the classification is SchemaClassification.UNKNOWN or SchemaClassification.UNSUPPORTED.
camt053.security.xml_guard.XmlSecurityError – For oversized / hostile payloads (DOCTYPE, ENTITY).
ValueError – For payloads that are not well-formed XML.

The dataclasses and exception types re-exported through this facade (AuditEvent, ChainVerification, HashChain, SchemaClassification, UnsupportedSchemaError) have their canonical documentation in camt053.audit and camt053.schema_version. They are intentionally excluded here so Sphinx does not emit duplicate-object-description warnings; the facade convenience (services.HashChain etc.) still works at runtime.

Audit log + schema-version negotiation#

The append-only HMAC hash-chain audit primitives (HashChain, AuditEvent, ChainVerification, verify_chain(), compute_event_hmac()) live in camt053.audit. Schema-version detection and classification (SchemaClassification, UnsupportedSchemaError, detect_schema_version(), classify_schema_version(), validate_schema_version()) lives in camt053.schema_version.

Both modules are re-exported through camt053.services for the facade convenience; the docstrings on the source modules carry the full reference text. Cross-references resolve to the canonical home.

Exceptions#

Every exception inherits from Camt053Error and carries a stable, machine-readable code.

Custom exception hierarchy for Camt053.

This module provides granular exception types to enable precise error handling in banking integrations. Instead of catching generic ValueError or TypeError, consuming applications can distinguish between data validation errors, configuration errors, and XML generation failures.

Every exception exposes a stable, machine-readable class-level code attribute. These codes are part of the public API: they will not change across releases and are guaranteed unique, so consumers can switch on them (e.g. to map a failure onto an HTTP status or a localised message) without relying on the exception’s class name or message text.

Error code taxonomy:

Code	Meaning
`CAMT053_ERROR`	Base error for any Camt053 failure.
`ACCOUNT_VALIDATION_ERROR`	Account/input data failed validation.
`XML_GENERATION_ERROR`	XML rendering or template failure.
`CONFIGURATION_ERROR`	Invalid configuration or CLI arguments.
`DATA_SOURCE_ERROR`	A data source could not be read.
`SCHEMA_VALIDATION_ERROR`	XML did not conform to its ISO 20022 XSD.
`INVALID_IBAN_ERROR`	IBAN format / checksum validation failed.
`INVALID_BIC_ERROR`	BIC/SWIFT format validation failed.
`INVALID_LEI_ERROR`	LEI format / checksum validation failed.
`MISSING_REQUIRED_FIELD_ERROR`	A required field was absent.
`STATEMENT_PARSE_ERROR`	An incoming statement could not be parsed.
`REVERSAL_GENERATION_ERROR`	A reversing entry could not be generated.

Example

>>> try:
...     process_files(...)
... except AccountValidationError as e:
...     # Handle invalid IBAN/BIC/LEI - notify user
...     log.error(f"[{e.code}] Account data invalid: {e}")
... except XMLGenerationError as e:
...     # Handle XML generation failure - check templates
...     log.error(f"[{e.code}] XML generation failed: {e}")
... except ConfigurationError as e:
...     # Handle config issues - check setup
...     log.error(f"[{e.code}] Configuration error: {e}")

exception camt053.exceptions.Camt053Error[source]

Bases: Exception

Base exception for all Camt053 errors.

All custom exceptions in this library inherit from this base class, allowing consumers to catch any Camt053-specific error with a single except clause if needed.

Example

>>> try:
...     process_files(...)
... except Camt053Error:
...     # Catch any Camt053-specific error
...     log.error("Camt053 operation failed")

code: str = 'CAMT053_ERROR'

exception camt053.exceptions.AccountValidationError(message, field=None)[source]

Bases: Camt053Error

Raised when account data validation fails.

This exception indicates issues with input data such as: - Invalid IBAN format - Invalid BIC/SWIFT code - Invalid LEI (Legal Entity Identifier) - Missing required fields (account owner, account servicer, etc.) - Invalid date or currency formats

Example

>>> try:
...     validate_account_data(data)
... except AccountValidationError as e:
...     # User-facing error - show validation message
...     return {"error": str(e), "field": e.field}

Parameters:

message (str)
field (str | None)

code: str = 'ACCOUNT_VALIDATION_ERROR'

exception camt053.exceptions.XMLGenerationError[source]

Bases: Camt053Error

Raised when XML generation or validation fails.

This exception indicates issues with: - Jinja2 template rendering failures - XSD schema validation errors - XML namespace issues - Missing or corrupted template files - Invalid XML structure

Example

>>> try:
...     generate_xml(data, template, schema)
... except XMLGenerationError as e:
...     # System error - check templates and schemas
...     log.error(f"XML generation failed: {e}")
...     alert_ops_team()

code: str = 'XML_GENERATION_ERROR'

exception camt053.exceptions.ConfigurationError[source]

Bases: Camt053Error

Raised when configuration or setup is invalid.

This exception indicates issues with: - Missing or invalid setup.cfg - Invalid CLI arguments - Missing required environment variables - Invalid file paths - Unsupported ISO 20022 version

Example

>>> try:
...     load_config("acmt.001.001.99")
... except ConfigurationError as e:
...     # Config error - show usage help
...     print(f"Configuration error: {e}")
...     print_usage_help()

code: str = 'CONFIGURATION_ERROR'

exception camt053.exceptions.DataSourceError[source]

Bases: Camt053Error

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_account_data("accounts.csv")
... except DataSourceError as e:
...     # Data access error - check file exists
...     log.error(f"Cannot access data source: {e}")

code: str = 'DATA_SOURCE_ERROR'

exception camt053.exceptions.SchemaValidationError(message, errors=None)[source]

Bases: Camt053Error

Raised when XSD schema validation fails.

This exception indicates issues with: - Generated XML does not conform to ISO 20022 schema - Missing required XML elements - Invalid XML element values - Namespace mismatches

Example

>>> try:
...     validate_xml_against_schema(xml, xsd)
... except SchemaValidationError as e:
...     # Schema validation error - check data mapping
...     log.error(f"XML schema validation failed: {e}")
...     log.debug(f"Validation errors: {e.errors}")

Parameters:

message (str)
errors (list[str] | None)

code: str = 'SCHEMA_VALIDATION_ERROR'

camt053.exceptions.XSDValidationError: alias of SchemaValidationError

exception camt053.exceptions.InvalidIBANError(message, iban, field=None, reason=None)[source]

Bases: AccountValidationError

Raised when IBAN validation fails.

This exception indicates issues with: - Invalid IBAN format (wrong structure) - Failed ISO 7064 mod-97-10 checksum validation - Unsupported country code - IBAN length mismatch for country

Example

>>> try:
...     validate_iban("AT68123456")  # Too short
... except InvalidIBANError as e:
...     # IBAN validation failed - prompt user to correct
...     print(f"Invalid IBAN: {e}")
...     print(f"Field: {e.field}, Value: {e.iban}")

Parameters:

message (str)
iban (str)
field (str | None)
reason (str | None)

code: str = 'INVALID_IBAN_ERROR'

exception camt053.exceptions.InvalidBICError(message, bic, field=None, reason=None)[source]

Bases: AccountValidationError

Raised when BIC/SWIFT validation fails.

This exception indicates issues with: - Invalid BIC format (must be 8 or 11 characters) - Invalid BIC structure (ISO 9362) - Invalid country code in BIC - Invalid bank/branch code characters

Example

>>> try:
...     validate_bic("INVALID123")
... except InvalidBICError as e:
...     # BIC validation failed - prompt user to correct
...     print(f"Invalid BIC: {e}")
...     print(f"Field: {e.field}, Value: {e.bic}")

Parameters:

message (str)
bic (str)
field (str | None)
reason (str | None)

code: str = 'INVALID_BIC_ERROR'

exception camt053.exceptions.InvalidLEIError(message, lei, field=None, reason=None)[source]

Bases: AccountValidationError

Raised when LEI (Legal Entity Identifier) validation fails.

This exception indicates issues with: - Invalid LEI format (must be 20 alphanumeric characters) - Failed ISO 17442 mod-97-10 checksum validation - Invalid structure (ISO 17442)

Example

>>> try:
...     validate_lei("INVALIDLEI")
... except InvalidLEIError as e:
...     # LEI validation failed - prompt user to correct
...     print(f"Invalid LEI: {e}")
...     print(f"Field: {e.field}, Value: {e.lei}")

Parameters:

message (str)
lei (str)
field (str | None)
reason (str | None)

code: str = 'INVALID_LEI_ERROR'

exception camt053.exceptions.MissingRequiredFieldError(message, field, row_number=None, required_fields=None)[source]

Bases: AccountValidationError

Raised when a required field is missing from account data.

This exception indicates issues with: - Missing mandatory fields (account_owner_name, account_servicer_bic, etc.) - Empty/null values for required fields - Missing fields in CSV rows - Missing dictionary keys

Example

>>> try:
...     validate_required_fields(data, ["account_owner_name", "msg_id"])
... except MissingRequiredFieldError as e:
...     # Required field missing - show user what's needed
...     print(f"Missing field: {e.field}")
...     print(f"Row: {e.row_number}, Expected fields: {e.required_fields}")

Parameters:

message (str)
field (str)
row_number (int | None)
required_fields (list[str] | None)

code: str = 'MISSING_REQUIRED_FIELD_ERROR'

exception camt053.exceptions.StatementParseError(message, line=None)[source]

Bases: Camt053Error

Raised when an incoming camt.05x statement cannot be parsed.

This exception indicates issues with: - Malformed or non-well-formed statement XML - A document whose root is not a recognised camt.05x message - Missing mandatory statement structure (group header, statement, entries)

Example

>>> try:
...     parse_document(xml)
... except StatementParseError as e:
...     # Bad inbound file - quarantine and alert operations
...     log.error(f"Cannot parse statement: {e}")

Parameters:

message (str)
line (int | None)

code: str = 'STATEMENT_PARSE_ERROR'

exception camt053.exceptions.ReversalGenerationError[source]

Bases: XMLGenerationError

Raised when a reversing entry cannot be generated.

This exception indicates issues with: - No statement entries matched the requested return/reason code - A booked entry missing the fields required to reverse it - Rendering or XSD validation failure of the reversal document

Example

>>> try:
...     generate_reversal(xml, reason_code="AC04")
... except ReversalGenerationError as e:
...     # Nothing to reverse, or reversal failed validation
...     log.error(f"Reversal generation failed: {e}")

code: str = 'REVERSAL_GENERATION_ERROR'