Examples#
These examples use synthetic data and the bundled templates. Their source
is included directly from scripts executed by tests/test_examples.py;
the manual does not maintain a second, untested copy of each program.
Run them from the repository root after
poetry install --all-extras --with dev.
See Verification coverage and limits for the feature-to-test evidence map and its limits. Passing an example does not certify a payment for a bank.
Example 1: Basic SEPA Payment Processing#
Load the bundled CSV, render XML, validate against the XSD and write to an explicit temporary output path. Scheme approval is a separate validation step; an ISO message edition alone does not identify a payment rail.
import tempfile
from pathlib import Path
from pain001.constants import TEMPLATES_DIR
from pain001.csv.load_csv_data import load_csv_data
from pain001.xml.generate_xml import generate_xml
MESSAGE_TYPE = "pain.001.001.03"
DATA_FILE = Path(__file__).resolve().parent / "data" / "payments.csv"
def main() -> None:
"""Load CSV data and write a validated pain.001 XML file."""
template_dir = TEMPLATES_DIR / MESSAGE_TYPE
data = load_csv_data(str(DATA_FILE))
print(f"Loaded {len(data)} payment rows from {DATA_FILE.name}")
with tempfile.TemporaryDirectory() as output_dir:
output_path = generate_xml(
data,
MESSAGE_TYPE,
str(template_dir / "template.xml"),
str(template_dir / f"{MESSAGE_TYPE}.xsd"),
output_path=str(Path(output_dir) / f"{MESSAGE_TYPE}.xml"),
)
size = Path(output_path).stat().st_size
print(f"Generated and XSD-validated {output_path} ({size} bytes)")
if __name__ == "__main__":
main()
Example 2: Batch Payment Processing from CSV#
Use the streaming workflow below for bounded chunks, or invoke the library
once per input with a distinct explicit output_path. A calculated filename
does not control output unless it is passed to the generation API. Never mark
an input successful merely because an exception was caught.
import tempfile
from pathlib import Path
from pain001.constants import TEMPLATES_DIR
from pain001.core.core import process_files_streaming
MESSAGE_TYPE = "pain.001.001.03"
TEMPLATE_DIR = TEMPLATES_DIR / MESSAGE_TYPE
def main() -> None:
"""Generate chunked XML files from the bundled sample data."""
with tempfile.TemporaryDirectory() as tmp:
written = process_files_streaming(
MESSAGE_TYPE,
str(TEMPLATE_DIR / "template.xml"),
str(TEMPLATE_DIR / f"{MESSAGE_TYPE}.xsd"),
str(TEMPLATE_DIR / "template.csv"),
chunk_size=2,
output_dir=tmp,
)
assert written, "expected at least one chunk file"
for path in written:
assert Path(path).exists()
print(f"Streaming wrote {len(written)} chunk file(s):")
for path in written:
print(f" {Path(path).name} ({Path(path).stat().st_size} bytes)")
print("Streaming example completed.")
if __name__ == "__main__":
main()
Example 3: Database-Driven Payment Processing#
The SQLite loader reads the configured payment table. Executing a separate
SQL query does not pass that query’s result to the loader. For a filtered
pipeline, pass the selected, normalized records to generate_xml_string;
test that selection separately. File generation is not bank acceptance and
must not by itself mark all pending ledger records as processed.
The runnable format example compares CSV, JSON, JSON Lines and Parquet rows and demonstrates reading the bundled SQLite database:
import json
import os
import shutil
import tempfile
from pathlib import Path
import pyarrow as pa
import pyarrow.parquet as pq
from pain001.constants import TEMPLATES_DIR
from pain001.csv.load_csv_data import load_csv_data
from pain001.data.loader import load_payment_data, load_payment_data_streaming
from pain001.db.load_db_data import load_db_data
TEMPLATE_DIR = TEMPLATES_DIR / "pain.001.001.03"
def main() -> None:
"""Load the bundled sample through CSV, SQLite, JSON, JSONL and Parquet."""
csv_rows = load_payment_data(str(TEMPLATE_DIR / "template.csv"))
print(f"CSV: {len(csv_rows)} rows")
db_rows = load_db_data(str(TEMPLATE_DIR / "template.db"), "pain001")
print(f"SQLite: {len(db_rows)} rows")
rows = load_csv_data(str(TEMPLATE_DIR / "template.csv"))
# The loader restricts paths to the working directory, so derive the
# JSON/JSONL files there rather than in the system temp dir.
work = tempfile.mkdtemp(dir=os.getcwd())
try:
json_file = Path(work) / "data.json"
json_file.write_text(json.dumps(rows))
json_rows = load_payment_data(str(json_file))
print(f"JSON: {len(json_rows)} rows")
jsonl_file = Path(work) / "data.jsonl"
jsonl_file.write_text("\n".join(json.dumps(r) for r in rows))
jsonl_rows = load_payment_data(str(jsonl_file))
print(f"JSONL: {len(jsonl_rows)} rows")
parquet_file = Path(work) / "data.parquet"
pq.write_table(pa.Table.from_pylist(rows), parquet_file)
parquet_rows = load_payment_data(str(parquet_file))
streamed = [
row
for batch in load_payment_data_streaming(
str(parquet_file), chunk_size=1
)
for row in batch
]
assert parquet_rows == streamed == rows
print(f"Parquet: {len(parquet_rows)} rows (eager and streaming agree)")
finally:
shutil.rmtree(work, ignore_errors=True)
# Every format yields the same number of payment rows.
assert csv_rows == json_rows == jsonl_rows == parquet_rows
assert db_rows
print("Input-formats example completed.")
if __name__ == "__main__":
main()
Example 4: Multi-Version Support#
Discover supported editions with pain001 versions. Choose the edition
and scheme required by the recipient; labels such as “instant” must not be
invented from an edition number. The regression matrix generates every
discovered bundled message type. This example demonstrates version mapping:
import tempfile
from pathlib import Path
from pain001.constants import TEMPLATES_DIR
from pain001.migration import VersionMapper
SOURCE = TEMPLATES_DIR / "pain.001.001.03" / "template.csv"
def main() -> None:
"""Migrate bundled v03 payment data to v09 and write the result."""
mapper = VersionMapper()
rows = mapper.migrate_file(
str(SOURCE), "pain.001.001.03", "pain.001.001.09"
)
mapper.validate_migrated_rows(rows, "pain.001.001.09")
assert rows and all(isinstance(r, dict) for r in rows)
print(f"Migrated {len(rows)} rows from v03 to v09")
with tempfile.TemporaryDirectory() as tmp:
out = Path(tmp) / "migrated_v09.csv"
mapper.write_csv(rows, str(out))
assert out.exists() and out.read_text().strip()
print(f"Wrote migrated CSV ({out.stat().st_size} bytes)")
print("Version-migration example completed.")
if __name__ == "__main__":
main()
Example 5: Error Handling and Logging#
The CLI example asserts success and failure exit codes. Logs and metrics
must not contain real account records or secrets. Use the actual exception
types documented in pain001.exceptions, not hypothetical exception
names or attributes. This executable metrics example shows application hooks:
import tempfile
from pathlib import Path
from pain001 import (
clear_metrics_callbacks,
process_files,
register_metrics_callback,
)
from pain001.constants import TEMPLATES_DIR
TEMPLATE_DIR = TEMPLATES_DIR / "pain.001.001.03"
def main() -> None:
"""Capture metric events emitted during a generation run."""
events = []
register_metrics_callback(lambda event: events.append(event))
try:
with tempfile.TemporaryDirectory() as tmp:
process_files(
"pain.001.001.03",
str(TEMPLATE_DIR / "template.xml"),
str(TEMPLATE_DIR / "pain.001.001.03.xsd"),
str(TEMPLATE_DIR / "template.csv"),
output_path=str(Path(tmp) / "out.xml"),
)
finally:
clear_metrics_callbacks()
assert events, "expected at least one metric event"
print(f"Captured {len(events)} metric event(s):")
for event in events[:8]:
print(f" {event.name} {event.attributes}")
print("Observability example completed.")
if __name__ == "__main__":
main()
Example 6: Large Batch Processing with Chunking#
Use process_files_streaming as demonstrated in Example 2. It preserves
record order and produces one validated output per chunk. Keep identifiers
as strings; generic dataframe type inference can destroy leading zeros.
The generation benchmark verifies output transaction and chunk counts in
addition to reporting memory and time. See Benchmark methodology.
Example 7: Payment Amount Validation#
Validate monetary values using exact decimal rules, not binary floating-point
comparisons or an invented Amount column. The policy example verifies the
limit boundary, warning behavior, request isolation and fail-closed handling
of missing inputs. Matching predicates produce findings; they do not modify
payment amounts. Schema and recipient-specific checks still apply.
from copy import deepcopy
from pain001.validation.policy import CustomRuleScheme, validate_policy
RULES = """
- id: REVIEW-LIMIT
where: 'payment_amount > decimal("50000.00")'
severity: error
message: Review the payment before approval.
"""
def main() -> None:
"""Verify exact thresholds, unchanged inputs and isolated rule evaluation."""
rows = [{"payment_amount": "50000.00"}, {"payment_amount": "50000.01"}]
original = deepcopy(rows)
result = CustomRuleScheme(RULES).validate(
rows, message_type="pain.001.001.03"
)
assert not result.is_valid
assert [(f.row_index, f.rule) for f in result.findings] == [
(1, "REVIEW-LIMIT")
]
assert rows == original
assert validate_policy(rows, RULES.replace("error", "warning")).is_valid
assert validate_policy(
rows, RULES.replace("50000.00", "60000.00")
).is_valid
try:
validate_policy([{}], RULES)
except ValueError as error:
assert "evaluation failed" in str(error)
else:
raise AssertionError("Missing policy fields must fail closed")
try:
CustomRuleScheme(RULES.replace('decimal("50000.00")', "50000.01"))
except ValueError as error:
assert "floating-point" in str(error)
else:
raise AssertionError("Inexact policy literals must be rejected")
print("CEL boundaries, warnings, isolation and fail-closed errors: passed")
if __name__ == "__main__":
main()
Example 8: Custom Data Processing Pipeline#
A plugin participates in the same eager and streaming dispatch as built-ins. This example proves that registration affects execution and preserves every row. It uses an example-only extension and no installed distribution changes. Production packaging and the pre-1.0 contract are documented in Plugin architecture.
from collections.abc import Iterator
from pathlib import Path
from tempfile import TemporaryDirectory
from pain001.constants import TEMPLATES_DIR
from pain001.csv.load_csv_data import load_csv_data
from pain001.data.loader import load_payment_data, load_payment_data_streaming
from pain001.plugins import (
PAIN001_API_VERSION,
LoaderResult,
PluginMeta,
registry,
)
class SyntheticLoader:
"""Treat an example-only extension as a CSV of synthetic payment rows."""
meta = PluginMeta(
"example-loader",
"0.0.1",
"Synthetic CSV",
api_version=PAIN001_API_VERSION,
)
extensions = (".example-payments",)
def load(self, source: str) -> LoaderResult:
"""Read the same validated fixture used by the built-in CSV loader."""
return LoaderResult(rows=load_csv_data(source))
def load_streaming(
self, source: str, chunk_size: int
) -> Iterator[LoaderResult]:
"""Yield bounded batches with the same order and content."""
rows = self.load(source).rows
for offset in range(0, len(rows), chunk_size):
yield LoaderResult(rows=rows[offset : offset + chunk_size])
def main() -> None:
"""Verify that registration changes execution, not just discovery."""
registry.list_plugins()
registry.register_loader(SyntheticLoader())
fixture = TEMPLATES_DIR / "pain.001.001.03" / "template.csv"
with TemporaryDirectory(dir=".") as directory:
source = Path(directory) / "payments.example-payments"
source.write_bytes(fixture.read_bytes())
eager = load_payment_data(str(source))
batches = list(load_payment_data_streaming(str(source), chunk_size=1))
assert eager == load_csv_data(str(fixture))
assert batches and all(len(batch) == 1 for batch in batches)
assert [row for batch in batches for row in batch] == eager
print("Registered loader: eager and streaming dispatch preserve all rows")
if __name__ == "__main__":
main()