pain001.validation package#

Validation module for Pain001.

This module provides centralized validation services for payment data, templates, schemas, and data sources.

Submodules#

pain001.validation.bic_validator module#

BIC (Business Identifier Code) / SWIFT Code validator.

This module implements ISO 9362 format validation for BIC/SWIFT codes. It validates format structure to catch typos before XML generation.

Example

>>> from pain001.validation.bic_validator import validate_bic
>>>
>>> # Valid BIC (8 characters)
>>> is_valid, error = validate_bic("DEUTDEFF")
>>> assert is_valid
>>>
>>> # Valid BIC (11 characters with branch code)
>>> is_valid, error = validate_bic("DEUTDEFF500")
>>> assert is_valid
>>>
>>> # Invalid BIC (wrong length)
>>> is_valid, error = validate_bic("DEUTDE")
>>> assert not is_valid

Standards:

ISO 9362:2022 - Banking — Banking telecommunication messages — Business identifier code (BIC)

pain001.validation.bic_validator.validate_bic(bic: str, field: str | None = None, strict: bool = True) → tuple[bool, str][source]#

Validate BIC/SWIFT code format.

This is the main entry point for BIC validation. It performs ISO 9362 format validation.

Parameters:

bic – BIC/SWIFT code to validate.
field – Optional field name for error reporting.
strict – If True, raise InvalidBICError on failure. If False, return tuple.

Returns:

Tuple of (is_valid, error_message). If strict=True and invalid, raises exception.

Raises:

InvalidBICError – If strict=True and BIC is invalid.

Example

>>> # Non-strict mode (returns tuple)
>>> is_valid, error = validate_bic("DEUTDEFF", strict=False)
>>> assert is_valid
>>>
>>> # Strict mode (raises exception on error)
>>> try:
...     validate_bic("INVALID", field="debtor_agent")
... except InvalidBICError as e:
...     print(f"Invalid: {e}")

pain001.validation.bic_validator.validate_bic_format(bic: str) → tuple[bool, str][source]#

Validate BIC format structure according to ISO 9362.

BIC Structure: - AAAABBCCXXX (11 characters) or AAAABBCC (8 characters) - AAAA: Bank code (4 letters, A-Z) - BB: Country code (2 letters, ISO 3166-1 alpha-2) - CC: Location code (2 alphanumeric, A-Z or 0-9) - XXX: Branch code (3 alphanumeric, optional)

Parameters:: bic – BIC/SWIFT code to validate (spaces removed automatically).
Returns:: Tuple of (is_valid, error_message).

Example

>>> is_valid, error = validate_bic_format("DEUTDEFF")
>>> assert is_valid
>>> is_valid, error = validate_bic_format("DEUTDEFF500")
>>> assert is_valid

pain001.validation.bic_validator.validate_bic_safe(bic: str, field: str | None = None) → bool[source]#

Validate BIC and return True/False (never raises exceptions).

This is a convenience wrapper for validate_bic with strict=False. Useful when you only need a boolean result without error details. When pain001-fast is available, delegates to fast structural native check.

Parameters:

bic – BIC/SWIFT code to validate.
field – Optional field name (unused, for API compatibility).

Returns:

True if BIC is valid, False otherwise.

Example

>>> if validate_bic_safe("DEUTDEFF"):
...     print("Valid BIC")

pain001.validation.charset module#

ISO 20022 restricted character-set validation and transliteration.

Most ISO 20022 text fields are limited to the “Latin character set” (also called the SWIFT-x / FIN-X set). Banks routinely reject messages that carry characters outside it — accented names, em dashes, currency symbols pasted from spreadsheets — even when the XML is otherwise XSD-valid. This module detects those characters and offers a best-effort transliteration to a compliant form.

The permitted set is:

a-z  A-Z  0-9  /  -  ?  :  (  )  .  ,  '  +  and space

pain001.validation.charset.ISO20022_ALLOWED_CHARACTERS = frozenset({' ', "'", '(', ')', '+', ',', '-', '.', '/', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ':', '?', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'})#: The ISO 20022 “Latin character set” permitted in restricted text fields.

pain001.validation.charset.find_invalid_characters(value: str) → list[str][source]#

Return the characters in value outside the ISO 20022 set.

Parameters:: value – The text to inspect.
Returns:: A sorted list of the unique disallowed characters. Empty when the value is fully compliant.

Example

>>> find_invalid_characters("Jose Muller & Co")
['&']

pain001.validation.charset.is_valid_charset(value: str) → bool[source]#

Report whether value uses only ISO 20022 permitted characters.

Parameters:: value – The text to check.
Returns:: True if every character is permitted, False otherwise.

Example

>>> is_valid_charset("Invoice 12345")
True
>>> is_valid_charset("Café")
False

pain001.validation.charset.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'

pain001.validation.corrections module#

Deterministic, review-only suggestions for non-financial record fields.

pain001.validation.corrections.suggest_record_fix(record: dict[str, Any], validation_error: dict[str, Any], message_type: str = 'pain.001.001.03') → dict[str, Any][source]#

Suggest one explicit patch without modifying the supplied record.

Schema bounds come from bundled schemas, never from an error supplied by an agent. Bank identifiers, amounts and currencies are refused even for whitespace-only changes. Every candidate requires human review and full revalidation; missing-field placeholders are not payment data. Unsupported message types propagate ValueError from SchemaValidator.

Parameters:

record – A single flat payment record, which remains unchanged.
validation_error – An object with field and rule keys.
message_type – Bundled schema supplying field types and length limits.

Returns:

An object containing patches and a cannot_autofix reason when no deterministic candidate is available.

pain001.validation.iban_validator module#

IBAN (International Bank Account Number) validator.

This module implements ISO 7064 mod-97-10 checksum validation for IBANs. It validates both format and checksum integrity before XML generation, significantly reducing bank rejection rates.

Example

>>> from pain001.validation.iban_validator import validate_iban
>>>
>>> # Valid IBAN
>>> is_valid, error = validate_iban("DE89370400440532013000")
>>> assert is_valid
>>>
>>> # Invalid checksum
>>> is_valid, error = validate_iban("DE00370400440532013000")
>>> assert not is_valid
>>> assert "checksum" in error.lower()

Standards:

ISO 13616:2020 - International Bank Account Number (IBAN)
ISO 7064 - Check digit mod-97-10 algorithm

pain001.validation.iban_validator.validate_iban(iban: str, field: str | None = None, strict: bool = True) → tuple[bool, str][source]#

Validate IBAN format and checksum.

This is the main entry point for IBAN validation. It performs both format validation and ISO 7064 mod-97-10 checksum verification.

Parameters:

iban – IBAN string to validate.
field – Optional field name for error reporting.
strict – If True, raise InvalidIBANError on failure. If False, return tuple.

Returns:

Tuple of (is_valid, error_message). If strict=True and invalid, raises exception.

Raises:

InvalidIBANError – If strict=True and IBAN is invalid.

Example

>>> # Non-strict mode (returns tuple)
>>> is_valid, error = validate_iban("DE89370400440532013000", strict=False)
>>> assert is_valid
>>>
>>> # Strict mode (raises exception on error)
>>> try:
...     validate_iban("DE00370400440532013000", field="debtor_account")
... except InvalidIBANError as e:
...     print(f"Invalid: {e}")

pain001.validation.iban_validator.validate_iban_checksum(iban: str) → tuple[bool, str][source]#

Validate IBAN checksum using ISO 7064 mod-97-10 algorithm.

Algorithm: 1. Move first 4 characters to end: “DE89370400440532013000” → “370400440532013000DE89” 2. Replace letters with numbers (A=10, B=11, …, Z=35) 3. Calculate mod 97 of resulting number 4. Valid IBAN has mod 97 = 1

Parameters:: iban – IBAN string to validate (with or without spaces).
Returns:: Tuple of (is_valid, error_message).

Example

>>> is_valid, error = validate_iban_checksum("DE89370400440532013000")
>>> assert is_valid

pain001.validation.iban_validator.validate_iban_format(iban: str) → tuple[bool, str][source]#

Validate IBAN format structure.

Checks: - Minimum 15 characters (shortest IBAN: Norway) - Maximum 34 characters (ISO 13616 limit) - First 2 characters are valid country code - Characters 3-4 are digits (check digits) - Remaining characters are alphanumeric - Length matches country-specific IBAN length

Parameters:: iban – IBAN string to validate (with or without spaces).
Returns:: Tuple of (is_valid, error_message).

Example

>>> is_valid, error = validate_iban_format("DE89370400440532013000")
>>> assert is_valid

pain001.validation.iban_validator.validate_iban_safe(iban: str, field: str | None = None) → bool[source]#

Validate IBAN and return True/False (never raises exceptions).

This is a convenience wrapper for validate_iban with strict=False. Useful when you only need a boolean result without error details. When pain001-fast is available, delegates to zero-allocation native mod-97.

Parameters:

iban – IBAN string to validate.
field – Optional field name (unused, for API compatibility).

Returns:

True if IBAN is valid, False otherwise.

Example

>>> if validate_iban_safe("DE89370400440532013000"):
...     print("Valid IBAN")

pain001.validation.policy module#

Bounded, request-local CEL policies implementing the scheme contract.

class pain001.validation.policy.CustomRuleScheme(source: str)[source]#

Bases: object

A policy instance scoped to one caller, never stored in a singleton.

Parameters:: source – Inline YAML containing the complete policy.
Variables:: meta – Plugin metadata; the instance is not globally registered.

meta = PluginMeta(name='custom', version='0.0.74', description='Apply bounded, deterministic CEL policy rules.', api_version=(0, 54), source='built-in')#

validate(rows: list[dict[str, Any]], *, message_type: str) → SchemeResult[source]#

Return structured findings for rules whose predicates are true.

Parameters:

rows – Flat payment records, never mutated.
message_type – Bundled schema used for numeric coercion.

Returns:

Findings in row and policy order; warnings do not reject a batch.

Raises:

ValueError – If inputs or expression evaluation are invalid.

pain001.validation.policy.country_of_iban(value: str) → StringType[source]#: Return an IBAN’s country only after validating the identifier.

pain001.validation.policy.day_of_week(value: str) → StringType[source]#: Return a stable English weekday for an ISO date.

pain001.validation.policy.decimal(value: str) → Decimal[source]#: Parse an exact, finite decimal threshold from a string.

pain001.validation.policy.is_business_day(value: str, country: str) → BoolType[source]#: Check weekdays and national public holidays, not bank settlement rules.

pain001.validation.policy.validate_policy(rows: list[dict[str, Any]], source: str, profile: str | None = None, message_type: str = 'pain.001.001.03') → SchemeValidationResult[source]#

Compose a private policy with the requested built-in scheme profiles.

Invalid policies, profiles or record evaluations propagate ValueError from the policy or scheme implementation.

Parameters:

rows – Flat payment records.
source – Inline YAML rules, never a server-side file path.
profile – Optional comma-separated scheme names, including custom.
message_type – The bundled message schema to validate against.

Returns:

The combined findings in stable row order.

pain001.validation.rails module#

Rail profiles: the public scheme rules of eighteen payment rails.

Each Rail is a declarative rulebook for one rail in one market, drawn from the public scheme and bank documents the corpus plan cites (ADR-0003, workstream 4): the currency and per-item ceiling, the service level and local instrument the rail expects, the shape of a domestic account and clearing member, reference lengths, purpose and regulatory-reporting mandates, address and UETR requirements, and for direct debits the mandate, sequence and creditor-identifier rules. One engine, RailProfile, turns a rail into a ValidationProfile, so every rail composes with the SEPA profiles and anti-duplicate through the comma syntax (--scheme uk-chaps,anti-duplicate) and reaches the CLI, REST and MCP surfaces through the plugin registry unchanged.

Rules a public document states are errors; rules the plan marks as typical bank practice or an assumption are warnings and say so. PurposeMandate adds the five country purpose-code mandates (AE, QA, MY, GB, HK) the plan names, as their own profiles.

pain001.validation.rails.BACS_DD_CODES: frozenset[str] = frozenset({'01', '17', '18', '19'})#: Bacs transaction codes the plan maps sequence types to (an assumption).

class pain001.validation.rails.Domestic(clearing_systems: tuple[str, ...], member_digits: int, account_digits: tuple[int, int], member_check: str | None = None)[source]#

Bases: object

A domestic routing-and-account shape.

Variables:

clearing_systems (tuple[str, ...]) – Accepted ClrSysId codes (GBDSC, USABA).
member_digits (int) – Length of the member id (sort code, routing number).
account_digits (tuple[int, int]) – Inclusive length range of the domestic account id.
member_check (str | None) – aba to verify the ABA check digit, else None.

account_digits: tuple[int, int]#

clearing_systems: tuple[str, ...]#

member_check: str | None = None#

member_digits: int#

class pain001.validation.rails.PurposeMandate(country: str, title: str, where: str, pattern: str | None = None, currency: str | None = None, sources: tuple[str, ...] = ())[source]#

Bases: ValidationProfile

A country’s purpose-code mandate, as its own profile.

Parameters:

country – ISO 3166 alpha-2 code.
title – Human title.
where – purpose (Purp/Cd or Prtry) or regulatory (RgltryRptg/Dtls), where the country expects the code.
pattern – Regex the code must match, if the country prescribes one.
currency – Only when the payment currency is this (HK RMB).
sources – The public documents the mandate comes from.

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the country’s purpose mandate.

Parameters:: data – Loaded payment rows.
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.rails.Rail(name: str, title: str, prefix: str, messages: tuple[str, ...], sources: tuple[str, ...], currencies: tuple[str, ...] = (), max_amount: Decimal | None = None, max_amount_severity: str = 'error', service_levels: tuple[str, ...] = (), local_instruments: tuple[str, ...] | None = None, local_instrument_proprietary: tuple[str, ...] = (), local_instrument_severity: str = 'warning', bic_required: bool = False, both_bics_required: bool = False, iban_countries: tuple[str, ...] = (), domestic: Domestic | None = None, end_to_end_max: int | None = None, remittance_max: int | None = None, remittance_severity: str = 'warning', purpose: str | None = None, regulatory: str | None = None, regulatory_pattern: Pattern[str] | None = None, charge_bearers: tuple[str, ...] = (), charge_bearer_severity: str = 'error', address_required: bool = False, address_severity: str = 'error', uetr_required: bool = False, sequence_types: tuple[str, ...] = (), mandate_max: int | None = None, creditor_id_pattern: Pattern[str] | None = None, creditor_reference: str | None = None, batch_booking: str | None = None)[source]#

Bases: object

One rail’s rulebook. Every field is optional; unset means unchecked.

Variables:

name (str) – Profile id, e.g. uk-fps.
title (str) – Human title.
prefix (str) – Rule-id prefix, e.g. UK-FPS.
messages (tuple[str, ...]) – Message families the rail carries.
sources (tuple[str, ...]) – The public documents the rules come from.
currencies (tuple[str, ...]) – Accepted currencies; empty means any ISO 4217 code.
max_amount (decimal.Decimal | None) – Per-item ceiling.
max_amount_severity (str) – error for a scheme limit, warning for a typical bank limit.
service_levels (tuple[str, ...]) – Expected SvcLvl/Cd values (warning when other).
local_instruments (tuple[str, ...] | None) – Accepted LclInstrm/Cd (empty tuple means the rail carries none; None means unchecked).
local_instrument_proprietary (tuple[str, ...]) – Accepted LclInstrm/Prtry values.
local_instrument_severity (str) – Severity of the local-instrument rule.
bic_required (bool) – The creditor agent (pain.008: debtor agent) needs a BIC.
both_bics_required (bool) – Both agents need a BIC.
iban_countries (tuple[str, ...]) – The counterparty IBAN must start with one of these.
domestic (pain001.validation.rails.Domestic | None) – Domestic routing shape, accepted instead of an IBAN.
end_to_end_max (int | None) – Ceiling on the end-to-end id length.
remittance_max (int | None) – Ceiling on unstructured remittance length.
remittance_severity (str) – Severity of the remittance rule.
purpose (str | None) – require, warn or None; a present purpose code must be in the ISO external list either way.
regulatory (str | None) – require or warn for a regulatory-reporting code.
regulatory_pattern (re.Pattern[str] | None) – Regex the regulatory info must match.
charge_bearers (tuple[str, ...]) – Accepted ChrgBr values.
charge_bearer_severity (str) – Severity of the charge-bearer rule.
address_required (bool) – Counterparty needs town and country.
address_severity (str) – Severity of the address rule.
uetr_required (bool) – The transaction needs a UETR (warning: .03 has none).
sequence_types (tuple[str, ...]) – Accepted SeqTp values (pain.008).
mandate_max (int | None) – Ceiling on the mandate id length (pain.008).
creditor_id_pattern (re.Pattern[str] | None) – Regex the creditor identifier must match.
creditor_reference (str | None) – ch (QRR/SCOR rules) or ocr (2 to 25 digits).
batch_booking (str | None) – Expected BtchBookg value (warning).

address_required: bool = False#

address_severity: str = 'error'#

batch_booking: str | None = None#

bic_required: bool = False#

both_bics_required: bool = False#

charge_bearer_severity: str = 'error'#

charge_bearers: tuple[str, ...] = ()#

creditor_id_pattern: Pattern[str] | None = None#

creditor_reference: str | None = None#

currencies: tuple[str, ...] = ()#

domestic: Domestic | None = None#

end_to_end_max: int | None = None#

iban_countries: tuple[str, ...] = ()#

local_instrument_proprietary: tuple[str, ...] = ()#

local_instrument_severity: str = 'warning'#

local_instruments: tuple[str, ...] | None = None#

mandate_max: int | None = None#

max_amount: Decimal | None = None#

max_amount_severity: str = 'error'#

messages: tuple[str, ...]#

name: str#

prefix: str#

purpose: str | None = None#

regulatory: str | None = None#

regulatory_pattern: Pattern[str] | None = None#

remittance_max: int | None = None#

remittance_severity: str = 'warning'#

sequence_types: tuple[str, ...] = ()#

service_levels: tuple[str, ...] = ()#

sources: tuple[str, ...]#

title: str#

uetr_required: bool = False#

class pain001.validation.rails.RailProfile(rail: Rail)[source]#

Bases: ValidationProfile

Validate rows against one Rail.

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the rail’s rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form, or the projection of a corpus file).
Returns:: A SchemeValidationResult listing every violation.

pain001.validation.rails.SEC_CODES: frozenset[str] = frozenset({'ACK', 'ADV', 'ARC', 'ATX', 'BOC', 'CCD', 'CIE', 'COR', 'CTX', 'DNE', 'ENR', 'IAT', 'MTE', 'POP', 'POS', 'PPD', 'RCK', 'SHR', 'TEL', 'TRC', 'TRX', 'WEB', 'XCK'})#: Nacha Standard Entry Class codes a pain.001 may carry in LclInstrm/Cd.

pain001.validation.rails.rail(name: str) → Rail[source]#

The Rail behind a profile name.

Parameters:: name – A rail profile id such as uk-fps.
Returns:: The rail.
Raises:: KeyError – If the name is not a rail.

pain001.validation.schema_validator module#

JSON Schema-based validator for payment data.

This module provides schema-based validation for payment data against ISO 20022 specifications. Schemas are externalized to JSON files to support multiple message types and custom validation rules.

Example

>>> from pain001.validation.schema_validator import SchemaValidator
>>>
>>> validator = SchemaValidator("pain.001.001.03")
>>> errors = validator.validate_data({
...     "id": "MSG001",
...     "date": "2024-01-15",
...     "nb_of_txs": 1,
...     # ... additional fields
... })
>>> if not errors:
...     print("✓ Data is valid")

class pain001.validation.schema_validator.SchemaValidator(message_type: str, schema_dir: Path | None = None)[source]#

Bases: object

Validates payment data against JSON Schema files.

The loaded schema dictionary is available as schema and the resolved schema file location as schema_path.

Parameters:

message_type – ISO 20022 message type (e.g., “pain.001.001.03”).
schema_dir – Directory containing schema files. Defaults to pain001/schemas/.

Raises:

ValueError – If the message type is not supported.
FileNotFoundError – If the schema file is missing or escapes the schema directory.
json.JSONDecodeError – If the schema file is invalid JSON.

get_field_description(field_name: str) → str | None[source]#

Get the description for a specific field.

Parameters:: field_name – Name of the field.
Returns:: Field description, or None if not available.

get_field_schema(field_name: str) → dict[str, Any] | None[source]#

Get the schema definition for a specific field.

Parameters:: field_name – Name of the field.
Returns:: Field schema dictionary, or None if field not in schema.

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

Extract required field names from schema.

Returns:: List of required field names.

validate_batch(rows: list[dict[str, Any]]) → tuple[int, int, list[tuple[int, list[ValidationError]]]][source]#

Validate a batch of rows.

Parameters:: rows – List of dictionaries containing payment data.
Returns:: Tuple of (total_rows, valid_rows, errors). errors is a list of (row_index, error_list) tuples for invalid rows.
Raises:: ValueError – If a registered validator returns an inconsistent validity flag or a finding outside the input batch.

validate_data(data: dict[str, Any]) → list[ValidationError][source]#

Validate a data dictionary against the schema.

String values are first coerced to the schema’s declared types so text-based sources (CSV, SQLite) validate the same as typed JSON.

Parameters:: data – Dictionary containing payment data to validate.
Returns:: List of ValidationError objects. Empty list if valid.
Raises:: ValueError – If the loaded JSON schema itself is invalid.

validate_row(row: dict[str, Any]) → tuple[bool, list[ValidationError]][source]#

Validate a single row of data.

Parameters:: row – Dictionary containing a single row of data.
Returns:: Tuple of (is_valid, errors). is_valid is True if no errors.

class pain001.validation.schema_validator.ValidationError(message: str, path: str, value: Any, rule: str)[source]#

Bases: object

Represents a validation error.

Parameters:

message – Human-readable error message.
path – JSON path to the invalid field (e.g., “$.debtor_account”).
value – The invalid value.
rule – The validation rule that failed (e.g., “pattern”, “required”).

pain001.validation.schemes module#

Stable scheme API and plugin dispatch, separate from pure built-in rules.

class pain001.validation.schemes.AntiDuplicateProfile(precision_overrides: Mapping[str, int] | None = None)#

Bases: ValidationProfile

Cross-record duplicate detection within a single batch.

Flags every group of two or more rows that share the same creditor IBAN, the same amount, and the same requested execution date: the signature of a payment that was keyed in (or exported) twice. Each row in a group is reported once, with the row indices of its partners in the message, so an operator can see both sides.

Unlike the SEPA and cross-border profiles this rulebook looks across rows rather than at each row on its own, which is why it is designed to compose: --scheme sepa-sct,anti-duplicate runs both and reports the union of findings.

Matching is exact-key only. Amounts are bucketed to the currency’s ISO 4217 minor unit (two places for EUR, none for JPY, three for KWD) so formatting noise does not hide a duplicate, but a one-cent difference is a different payment. Rows missing any key field are skipped; the intra-record profiles report those. The engine has no memory between runs, so cross-batch deduplication is out of scope.

Variables:

name (str) – Stable profile identifier, "anti-duplicate".
rule (str) – Rule id raised for every row in a duplicate group, "DUP-CREDITOR-DATE".

Parameters:

precision_overrides – Optional mapping of ISO 4217 code to the number of decimal places used to bucket amounts in that currency, overriding the built-in minor-unit table.

name: str = 'anti-duplicate'#

rule: str = 'DUP-CREDITOR-DATE'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Group rows by their duplicate key and flag every group of two+.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult with one DUP-CREDITOR-DATE violation per row that has at least one duplicate, in row order.

class pain001.validation.schemes.CrossBorderCreditTransferProfile#

Bases: ValidationProfile

Generic cross-border credit transfer checks (beyond SEPA).

A pragmatic, multi-currency rulebook for international credit transfers: valid debtor/creditor IBANs, a well-formed ISO 4217 currency (any currency, not only EUR), a mandatory creditor agent BIC (cross-border routing needs it), a positive amount within the ISO ceiling, and ISO 20022 charset/length limits. It is intentionally generic rather than a full CBPR+ implementation.

name: str = 'xborder-ct'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the cross-border rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.schemes.SchemeValidationResult(profile: str, violations: list[~pain001.validation.schemes.SchemeViolation] = <factory>)#

Bases: object

Outcome of validating payment rows against a scheme profile.

Variables:

profile (str) – Name of the profile that produced this result.
violations (list[pain001.validation.schemes.SchemeViolation]) – All violations found, in row order.

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.validation.schemes.SchemeViolation(rule: str, message: str, index: int, field: str | None = None, severity: str = 'error')#

Bases: object

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

Variables:

rule (str) – Stable identifier of the rule (e.g. "SEPA-CCY").
message (str) – Human-readable description of the breach.
index (int) – Zero-based index of the offending payment row.
field (str | None) – Name of the offending field, when applicable.
severity (str) – "error" (scheme would reject) or "warning".

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.validation.schemes.SepaB2BDirectDebitProfile#

Bases: ValidationProfile

SEPA Business-to-Business Direct Debit (B2B) rulebook checks.

Adds the B2B-specific constraints - a creditor identifier and the stricter B2B sequence-type set (FRST / RCUR only) - on top of the shared SDD rules (mandate id, EUR, IBANs, BIC, amount ceiling, charset, length). Use this profile when collecting from corporate debtors under a B2B mandate; the consumer CORE rules (sepa-sdd) are too permissive for B2B.

name: str = 'sepa-b2b'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the SEPA B2B SDD rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.schemes.SepaCreditTransferProfile#

Bases: ValidationProfile

SEPA Credit Transfer (SCT) rulebook checks.

Enforces the core, machine-checkable SCT constraints that the XSD cannot: EUR currency, valid debtor/creditor IBANs, optional but well-formed BIC, the EUR amount ceiling, and ISO 20022 character-set and length limits on text fields.

name: str = 'sepa-sct'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the SEPA SCT rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.schemes.SepaDirectDebitProfile#

Bases: ValidationProfile

SEPA Direct Debit (SDD) rulebook checks.

Adds the direct-debit-specific constraints — a mandate reference and a valid sequence type — on top of the shared SEPA rules (EUR currency, IBANs, BIC, amount ceiling, charset, and length).

name: str = 'sepa-sdd'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the SEPA SDD rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.schemes.SepaInstantCreditTransferProfile#

Bases: ValidationProfile

SEPA Instant Credit Transfer (SCT Inst) rulebook checks.

Identical to SEPA SCT but enforces the instant scheme’s stricter per-transaction ceiling of 100,000.00 EUR (rule SEPA-INST-AMT).

name: str = 'sepa-inst'#

validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against the SEPA SCT Inst rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

class pain001.validation.schemes.ValidationProfile#

Bases: ABC

Base class for a payment-scheme rulebook validator.

name: str = ''#

abstract validate(data: list[dict[str, Any]]) → SchemeValidationResult[source]#

Validate payment rows against this scheme’s rulebook.

Parameters:: data – Loaded payment rows (the normalised internal form).
Returns:: A SchemeValidationResult listing every violation.

pain001.validation.schemes.remediation_for(rule: str) → str#

Return the remediation hint for a rule id.

Parameters:: rule – The rule identifier (e.g. "SEPA-CCY").
Returns:: The remediation hint, or an empty string if the rule is unknown.

pain001.validation.schemes.validate_scheme(data: list[dict[str, Any]], profile: str = 'sepa-sct', *, message_type: str = 'pain.001.001.03') → 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"), or a comma-separated list of names ("sepa-sct,anti-duplicate") to run several rulebooks and report the union of their findings. The combined result’s profile joins the names with commas and its violations are ordered by row.
message_type – Message type passed to registered scheme plugins.

Returns:

A SchemeValidationResult listing every violation.

Raises:

ValueError – If profile names a profile that is not registered.

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'
>>> both = validate_scheme(rows * 2, profile="sepa-sct,anti-duplicate")
>>> both.profile
'sepa-sct,anti-duplicate'

pain001.validation.service module#

ValidationService: Centralized validation orchestrator for Pain001.

This module decouples validation logic from the CLI, enabling validation to be invoked programmatically from API endpoints, web hooks, and batch jobs.

Example

>>> from pain001.validation import ValidationService, ValidationConfig
>>>
>>> config = ValidationConfig(
...     xml_message_type="pain.001.001.03",
...     xml_template_file_path="template.xml",
...     xsd_schema_file_path="schema.xsd",
...     data_file_path="payments.csv"
... )
>>>
>>> service = ValidationService()
>>> report = service.validate_all(config)
>>>
>>> if not report.is_valid:
...     for error in report.errors:
...         print(f"Validation error: {error}")

class pain001.validation.service.ValidationConfig(xml_message_type: str, xml_template_file_path: str, xsd_schema_file_path: str, data_file_path: str, pre_validate: bool = True)[source]#

Bases: object

Configuration for validation operations.

Variables:

xml_message_type (str) – ISO 20022 message type (e.g., ‘pain.001.001.03’).
xml_template_file_path (str) – Path to Jinja2 XML template file.
xsd_schema_file_path (str) – Path to XSD schema for validation.
data_file_path (str) – Path to CSV or SQLite data file.
pre_validate (bool) – Whether to run pre-validation (IBAN/BIC checksums).

data_file_path: str#

pre_validate: bool = True#

xml_message_type: str#

xml_template_file_path: str#

xsd_schema_file_path: str#

class pain001.validation.service.ValidationReport(is_valid: bool, errors: list[str] = <factory>, results: dict[str, ~pain001.validation.service.ValidationResult] = <factory>)[source]#

Bases: object

Comprehensive validation report for all checks.

Variables:

is_valid (bool) – Whether all validations passed.
errors (list[str]) – List of error messages from failed validations.
results (dict[str, pain001.validation.service.ValidationResult]) – Dictionary of individual validation results.

errors: list[str]#

is_valid: bool#

results: dict[str, ValidationResult]#

class pain001.validation.service.ValidationResult(is_valid: bool, error: str | None = None, field: str | None = None, details: str | None = None)[source]#

Bases: object

Result of a single validation check.

Variables:

is_valid (bool) – Whether the validation passed.
error (str | None) – Optional error message if validation failed.
field (str | None) – Optional field name that caused the validation error.
details (str | None) – Optional additional details about the validation.

details: str | None = None#

error: str | None = None#

field: str | None = None#

is_valid: bool#

class pain001.validation.service.ValidationService[source]#

Bases: object

Centralized validation orchestrator for Pain001.

This service provides programmatic access to validation logic, decoupled from the CLI interface. It can be invoked from API endpoints, web hooks, batch jobs, or directly from Python code.

Example

>>> service = ValidationService()
>>> result = service.validate_template(Path("template.xml"))
>>> if result.is_valid:
...     print("Template is valid")
... else:
...     print(f"Template error: {result.error}")

validate_all(config: ValidationConfig) → ValidationReport[source]#

Run comprehensive validation pipeline.

This method runs all validation checks and returns a consolidated report. It’s the primary entry point for validation operations.

Parameters:: config – Validation configuration with all required paths.
Returns:: ValidationReport with results from all checks.
Raises:: ConfigurationError – If config is invalid.

Example

>>> service = ValidationService()
>>> config = ValidationConfig(
...     xml_message_type="pain.001.001.03",
...     xml_template_file_path="template.xml",
...     xsd_schema_file_path="schema.xsd",
...     data_file_path="payments.csv"
... )
>>> report = service.validate_all(config)
>>> if not report.is_valid:
...     for error in report.errors:
...         print(f"Error: {error}")

validate_data_content(data_path: str | Path) → ValidationResult[source]#

Validate that the data file can be loaded and parsed.

Parameters:: data_path – Path to CSV or SQLite data file.
Returns:: ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_data_content("payments.csv")
>>> if not result.is_valid:
...     print(f"Data validation failed: {result.error}")

validate_data_source(data_path: str | Path) → ValidationResult[source]#

Validate that the data source file exists and is accessible.

Parameters:: data_path – Path to CSV or SQLite data file.
Returns:: ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_data_source("payments.csv")
>>> if not result.is_valid:
...     print(f"Data file not found: {result.error}")

validate_message_type(message_type: str) → ValidationResult[source]#

Validate that the message type is supported.

Parameters:: message_type – ISO 20022 message type (e.g., ‘pain.001.001.03’).
Returns:: ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_message_type("pain.001.001.03")
>>> assert result.is_valid

validate_schema(schema_path: str | Path) → ValidationResult[source]#

Validate that the XSD schema file exists and is accessible.

Parameters:: schema_path – Path to XSD schema file.
Returns:: ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_schema("schema.xsd")
>>> if not result.is_valid:
...     print(f"Schema not found: {result.error}")

validate_template(template_path: str | Path) → ValidationResult[source]#

Validate that the XML template file exists and is accessible.

Parameters:: template_path – Path to XML template file.
Returns:: ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_template("template.xml")
>>> if not result.is_valid:
...     print(f"Template not found: {result.error}")

validate_template_schema_compatibility(template_path: str | Path, schema_path: str | Path) → ValidationResult[source]#

Validate that the XML template conforms to the XSD schema.

Note

Raw Jinja2 XML templates containing unrendered directives ({% for … %}, {{ date }}) are not strictly schema-valid XML until rendered with payment data. Schema validation is therefore authoritatively enforced during document generation.

Parameters:

template_path – Path to XML template file.
schema_path – Path to XSD schema file.

Returns:

ValidationResult indicating success or failure.

Example

>>> service = ValidationService()
>>> result = service.validate_template_schema_compatibility(
...     "template.xml", "schema.xsd"
... )
>>> if not result.is_valid:
...     print(f"Schema validation failed: {result.error}")