pain001.plugins package#

Public plugin contract for pain001 v0.0.54+.

Four PEP 544 Protocol types define the surface any external package can implement to extend pain001 without forking:

AbstractLoader - read payment rows from a file format (CSV, XLSX, encrypted .gpg, etc.).
AbstractValidator - validate parsed rows in isolation (schema, types, identifier formats).
AbstractScheme - validate parsed rows against a payment-scheme rulebook (SEPA SCT, SDD, instant, custom CEL).
AbstractWriter - serialise generator output (currently XML; future: SWIFT MX, JSON, base64-wrapped).

A plugin package is published to PyPI with one or more entry points in the pain001.loaders / .validators / .schemes / .writers groups (declared in its pyproject.toml). pain001 discovers them at process start via importlib.metadata.entry_points and registers each in the process-level pain001.plugins.registry.

Compatibility#

Every plugin should declare PAIN001_API_VERSION = (major, minor) as a class attribute. The registry warns (but loads) when the plugin’s declared minor is lower than the running pain001 minor and refuses to load when the major is higher. This is a stable promise: within a major version we never break a plugin that compiles against the published Protocols.

See docs/plugins.md for the worked example and the contract versioning policy.

Submodules#

pain001.plugins.builtins_gpg module#

Built-in GPG-decrypting loader (requires the pain001[gpg] extra).

A composition plugin: it does not understand the inner payload format itself; instead it decrypts foo.csv.gpg (or .asc) in memory, writes the plaintext to a brief, 0o600-permissioned temp file in /dev/shm (Linux RAM-backed tmpfs) or the platform default otherwise, dispatches to the loader registered for the inner extension (.csv, .json, …), and unlinks the temp file before returning.

Plaintext therefore touches the filesystem for the duration of the inner loader’s read - on Linux that is a RAM-backed file, on other platforms an unlinked temp file with restrictive permissions. This trade-off is documented in docs/plugins.md; a future loader contract version may add a bytes-native API to eliminate the temp file entirely.

Configuration is environment-driven (the same surface ops teams use for the REST API and Redis backends):

PAIN001_GPG_HOMEDIR - GPG homedir (default: ~/.gnupg).
PAIN001_GPG_PASSPHRASE_ENV - name of the env var that holds the private-key passphrase, if the key is passphrase-protected.
PAIN001_GPG_KEYFILE - optional private-key file imported into the homedir before decrypting (the CLI --decrypt-key flag sets this). Use it when the key lives in a secret store rather than a keyring, e.g. in CI.
PAIN001_GPG_KEY_ID - optional specific key id to decrypt with; when unset, any matching key in the keyring is tried.

The CLI exposes the two most common knobs as flags: --decrypt-key <keyfile> and --decrypt-passphrase-env <VAR>.

exception pain001.plugins.builtins_gpg.GpgDecryptError[source]#

Bases: DataSourceError

Raised when GPG decryption fails for any reason.

Wraps the underlying gnupg error so callers see a single pain001-typed exception, with no key material or stack trace leaking into the message.

pain001.plugins.builtins_gpg.maybe_register(reg: BuiltinRegistry) → None[source]#

Register the GPG loader iff the python-gnupg extra is installed.

Called by pain001.plugins._builtins.register_all so the GPG loader is opt-in via the extra rather than always-on.

Parameters:: reg – The process-level PluginRegistry to populate.

pain001.plugins.companions module#

Install hints for formats handled by companion packages.

pain001 dispatches unknown extensions to the plugin registry, so a format it does not bundle is not an error in itself — it means the plugin that handles it is not installed. Saying so generically (“install a plugin that registers ‘.xlsx’”) leaves the user to work out which plugin, which is the part they do not know.

This module is the single place that maps an extension to the package that handles it. It holds no import of the companion and no dependency on it: it exists purely so the error message can name the thing to install.

Keep it small. An entry belongs here only when the package is published, first-party, and the sole obvious handler for that extension — otherwise the hint becomes an advert, or worse, points at something abandoned.

pain001.plugins.companions.COMPANION_LOADERS: Final[mappingproxy[str, str]] = mappingproxy({'.xlsx': 'pain001-loader-xlsx', '.xlsm': 'pain001-loader-xlsx', '.gpg': 'pain001[gpg]', '.asc': 'pain001[gpg]'})#

Extension -> the pip target that registers a loader for it.

Read-only so a caller cannot reshape a process-wide table while composing an error message.

pain001.plugins.companions.install_hint(extension: str) → str | None[source]#

Return the pip target that handles extension, if one is known.

Parameters:: extension – File extension including the leading dot. Matched case-insensitively, because Windows exports routinely produce .XLSX.
Returns:: The pip install target ("pain001-loader-xlsx"), or None when no first-party package claims the extension.

Example

>>> install_hint(".XLSX")
'pain001-loader-xlsx'
>>> install_hint(".rtf") is None
True

pain001.plugins.contracts module#

Protocol definitions for the pain001 plugin contract.

Every protocol here is the minimum surface a plugin must satisfy for pain001 to dispatch to it. They are intentionally narrow:

No mandatory __init__ shape (plugins are instantiated by the registry with no arguments, so any plugin needing config reads it from environment variables or pain001.config).
No exceptions are listed as part of the protocol; plugins should raise pain001.exceptions.DataSourceError, pain001.exceptions.PaymentValidationError, or any subclass thereof. Anything else is treated as a plugin bug.
All payment data passes through pain001 as list[dict[str, Any]] (flat records); plugins must respect that shape and not introduce their own envelopes.

Forward compatibility: when the contract grows a method, it is added as an optional method with a default that calls the existing surface. We never remove a method inside the same major.

class pain001.plugins.contracts.AbstractLoader(*args, **kwargs)[source]#

Bases: Protocol

A reader that turns a file path into list[dict] payment rows.

Loaders register against one or more file extensions; the registry dispatches by suffix at parse time. Required attributes: meta (a PluginMeta) and extensions (a tuple of file extensions with leading dot, lower-case, that this loader handles - (".xlsx", ".xlsm") etc.).

extensions: tuple[str, ...]#

load(path: str) → LoaderResult[source]#

Read every row from path and return them all at once.

Parameters:: path – Absolute or relative filesystem path.
Returns:: A LoaderResult carrying the parsed rows and a source hint (typically the input path).

load_streaming(path: str, chunk_size: int) → Iterable[LoaderResult][source]#

Return an iterable of LoaderResult chunks.

Required for --streaming mode. Loaders that cannot stream (e.g. small inline JSON) may return a single-element iterable holding every row.

Parameters:

path – Absolute or relative filesystem path.
chunk_size – Maximum rows per chunk.

Returns:

An iterable of LoaderResult instances. The caller treats each as a self-contained sub-batch.

meta: PluginMeta#

class pain001.plugins.contracts.AbstractScheme(*args, **kwargs)[source]#

Bases: Protocol

A whole-batch rulebook validator (sees every row at once).

Use this when a rule needs cross-record context (duplicate detection, batch totals, BIC reachability against a reference list). For row-by-row checks, prefer AbstractValidator. Required attribute: meta (a PluginMeta).

meta: PluginMeta#

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

Apply the rulebook to the full batch.

Parameters:

rows – Full batch produced by the loader (and possibly pre-filtered by upstream validators). Implementations must not mutate this list.
message_type – ISO 20022 message type the scheme is being run against; some rules vary by message type.

Returns:

A SchemeResult aggregating findings.

class pain001.plugins.contracts.AbstractValidator(*args, **kwargs)[source]#

Bases: Protocol

An intra-record validator (one row at a time, no batch context).

Required attribute: meta (a PluginMeta).

meta: PluginMeta#

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

Validate every row independently and return the aggregate.

Parameters:

rows – The full batch the loader produced. Implementations must not mutate this list.
message_type – ISO 20022 message type the validation should target (e.g. "pain.001.001.09").

Returns:

A ValidatorResult aggregating per-row findings.

class pain001.plugins.contracts.AbstractWriter(*args, **kwargs)[source]#

Bases: Protocol

A serialiser that takes the rendered XML and writes it somewhere.

Default writer writes to a filesystem path; future writers may upload to SFTP, push to S3, base64-wrap into a JSON envelope, etc. The pivot from “what to render” to “where it goes” lives here. Required attribute: meta (a PluginMeta).

meta: PluginMeta#

write(xml: str, destination: str) → str[source]#

Write xml to destination and return the canonical sink.

Parameters:

xml – The validated ISO 20022 XML document, as a string. Writers must not re-parse, re-serialise, or alter the XML in any way (canonical form is decided by the generator).
destination – Writer-specific location string. The default file writer treats this as a filesystem path; an SFTP writer would accept sftp://user@host/path; etc.

Returns:

A canonical string identifying where the bytes ended up (an absolute path, an sftp:// URL, an s3:// URI). The caller logs this and returns it from the public API.

class pain001.plugins.contracts.LoaderResult(rows: list[dict[str, Any]], source_hint: str = '<inline>')[source]#

Bases: object

The two pieces a loader yields on success.

rows carries the parsed flat records, one per payment, ready for downstream validation and rendering. source_hint is a diagnostic string (typically the original file path or "<inline>") that pain001 attaches to validation findings so the user can locate the offending row in the source.

rows: list[dict[str, Any]]#

source_hint: str = '<inline>'#

class pain001.plugins.contracts.PluginInfo(kind: str, meta: PluginMeta)[source]#

Bases: object

Lightweight record returned by PluginRegistry.list_* calls.

A flattened view of the meta + plugin kind, used by the CLI and JSON-RPC inspection tools rather than the live plugin object. kind is one of "loader", "validator", "scheme", "writer"; meta is the plugin’s declared PluginMeta.

kind: str#

meta: PluginMeta#

class pain001.plugins.contracts.PluginMeta(name: str, version: str, description: str, api_version: tuple[int, int] = (0, 54), source: str = 'built-in')[source]#

Bases: object

Descriptive metadata every plugin must expose.

Stored on the plugin class as the meta attribute, surfaced by pain001 plugins list so an operator can audit what the running interpreter has loaded. Field semantics:

name - stable, kebab-case identifier ("csv", "sepa-sct"). Must be unique within its plugin group; a third-party plugin with the same name as a built-in wins.
version - SemVer string of the package shipping the plugin (importlib.metadata.version(...)).
description - one-line human-readable summary; shown by the CLI. Keep under 80 characters.
api_version - (major, minor) of the pain001 plugin contract the plugin targets. Defaults to the current value.
source - "built-in" for plugins shipped inside pain001; "<dist-name>==<version>" for third-party plugins. Auto-filled by the registry; plugins must not set it.

api_version: tuple[int, int] = (0, 54)#

description: str#

name: str#

source: str = 'built-in'#

version: str#

class pain001.plugins.contracts.SchemeFinding(row_index: int, field: str | None, rule: str, severity: str, message: str, related_rows: tuple[int, ...] = (), remediation: str | None = None)[source]#

Bases: object

A single scheme-rulebook violation.

Same shape as ValidatorFinding plus related_rows (zero-based indices of other rows the rule relates to, for cross-record findings) and an optional remediation hint surfaced by --explain. row_index is the zero-based primary row index; field is a row field or None for batch-level findings; rule is a stable kebab-case rule id ("SEPA-CCY", "DUP-CREDITOR-DATE"); severity is one of "error" / "warning" / "info"; message ends in a period.

field: str | None#

message: str#

related_rows: tuple[int, ...] = ()#

remediation: str | None = None#

row_index: int#

rule: str#

severity: str#

class pain001.plugins.contracts.SchemeResult(is_valid: bool, findings: list[~pain001.plugins.contracts.SchemeFinding] = <factory>)[source]#

Bases: object

Outcome of one scheme running against one batch.

is_valid is True only when findings has zero error-severity entries; findings is the list produced by the scheme.

findings: list[SchemeFinding]#

is_valid: bool#

class pain001.plugins.contracts.ValidatorFinding(row_index: int, field: str | None, rule: str, severity: str, message: str)[source]#

Bases: object

A single row-level validation failure.

Field semantics: row_index is the zero-based index in the input batch where the failure occurred; field is the field name on the row ("debtor_account_IBAN") or None for row-level findings; rule is the stable kebab-case identifier of the rule that failed ("IBAN-CHECKSUM"); severity is one of "error", "warning", "info"; message is a human-readable explanation ending in a period.

field: str | None#

message: str#

row_index: int#

rule: str#

severity: str#

class pain001.plugins.contracts.ValidatorResult(is_valid: bool, findings: list[~pain001.plugins.contracts.ValidatorFinding] = <factory>)[source]#

Bases: object

Outcome of one validator running against one batch.

is_valid is True only when findings contains zero error-severity entries (warnings do not invalidate). findings is the full list produced by the validator.

findings: list[ValidatorFinding]#

is_valid: bool#

pain001.plugins.registry module#

Process-level plugin registry + entry-point discovery.

Holds one PluginRegistry instance per Python process. The registry is populated lazily on first use:

Built-in plugins shipped inside pain001 register first.
External plugins discovered via importlib.metadata.entry_points register on top, winning over built-ins with the same name.
The PAIN001_DISABLE_PLUGINS env var (comma-separated names) skips matching plugins regardless of source - the production escape hatch when a plugin breaks.

Plugin authors talk to the registry only through pain001.plugins.registry.registry() (the module-level singleton); inside pain001, the loader / validator / scheme dispatchers also go through the same singleton so external plugins and built-ins share one code path.

class pain001.plugins.registry.PluginRegistry(api_version: tuple[int, int])[source]#

Bases: object

In-process registry of every loaded plugin, keyed by kind + name.

Not thread-safe for registration (plugins are registered once at startup and the registry is read-only after); fully safe for concurrent lookups thereafter.

Parameters:: api_version – The (major, minor) tuple the host pain001 build implements. Plugins declaring a higher major are refused; plugins declaring a higher minor are loaded with a warning.

get_loader(name: str) → AbstractLoader | None[source]#: Return the loader registered under name (or None).

get_loader_for_extension(extension: str) → AbstractLoader | None[source]#

Return the loader registered for extension (with leading dot).

Parameters:: extension – File extension to dispatch on, with the leading dot, case-insensitive (".xlsx", ".XLSX" both work).
Returns:: The matching AbstractLoader, or None if none is registered.

get_scheme(name: str) → AbstractScheme | None[source]#: Return the scheme registered under name (or None).

get_validator(name: str) → AbstractValidator | None[source]#: Return the validator registered under name (or None).

get_writer(name: str) → AbstractWriter | None[source]#: Return the writer registered under name (or None).

list_plugins(kind: str | None = None) → list[PluginInfo][source]#

Return a flat list of every registered plugin’s metadata.

Parameters:: kind – Optional filter ("loader", "validator", "scheme", "writer"). When None, every kind is included.
Returns:: A list of PluginInfo, sorted by (kind, name).

register_loader(loader: AbstractLoader) → None[source]#: Register loader under its declared name; later wins.

register_scheme(scheme: AbstractScheme) → None[source]#: Register scheme under its declared name; later wins.

register_validator(validator: AbstractValidator) → None[source]#: Register validator under its declared name; later wins.

register_writer(writer: AbstractWriter) → None[source]#: Register writer under its declared name; later wins.

reset() → None[source]#

Drop every plugin and mark the registry unpopulated again.

Test-only seam. Production code never resets the singleton.

exception pain001.plugins.registry.PluginRegistryError[source]#

Bases: RuntimeError

Raised when the registry cannot honour a lookup or registration.

exception pain001.plugins.registry.PluginVersionError[source]#

Bases: PluginRegistryError

Raised when a plugin declares an API version newer than this build.

pain001.plugins.registry.registry: PluginRegistry = <pain001.plugins.registry.PluginRegistry object>#: Process-level singleton; everyone reads through this one.