pain001.api package

Pain001 FastAPI REST API module.

Submodules

pain001.api.app module

Pain001 FastAPI application.

async pain001.api.app.cancel_job(job_id: str) → dict[str, str]

Cancel an async job.

Parameters:

job_id – Job identifier.

Returns:

Dictionary with cancellation status.

Raises:

HTTPException – If job not found.

async pain001.api.app.download_xml(job_id: str) → FileResponse

Download generated XML file.

Parameters:

job_id – Job identifier.

Returns:

XML file for download.

Return type:

FileResponse

Raises:

HTTPException – If job not found or file not available.

async pain001.api.app.generate_xml_async(request: GenerateXMLRequest) → dict[str, str]

Start async XML generation job.

Parameters:

request – Generation request.

Returns:

Dictionary with job_id for status polling.

Raises:

HTTPException – If job creation fails.

async pain001.api.app.generate_xml_sync(request: GenerateXMLRequest) → GenerateXMLResponse

Generate XML synchronously.

Parameters:

request – Generation request with data source and options.

Returns:

Generated XML file path or errors.

Return type:

GenerateXMLResponse

Raises:

HTTPException – If generation fails.

async pain001.api.app.get_job_status(job_id: str) → JobStatusResponse

Get status of async job.

Parameters:

job_id – Job identifier.

Returns:

Current job status and result.

Return type:

JobStatusResponse

Raises:

HTTPException – If job not found.

async pain001.api.app.health() → HealthResponse

Check API health status.

Returns:

API status and version information.

Return type:

HealthResponse

async pain001.api.app.metrics() → PlainTextResponse

Expose Prometheus metrics (build info, gauges, request counters).

Returns:

The metrics document in Prometheus text exposition format.

async pain001.api.app.scalar_reference() → HTMLResponse

Serve an interactive Scalar API reference for the OpenAPI document.

Returns:

The Scalar reference UI as an HTML response.

async pain001.api.app.validate_data(request: ValidationRequest) → ValidationResponse

Validate payment data against schema.

Parameters:

request – Validation request with data source and file path.

Returns:

Validation results with error details.

Return type:

ValidationResponse

Raises:

HTTPException – If file not found or validation fails.

pain001.api.auth module

Bearer-token authentication shared by every REST router.

Lives in its own module so the dashboard router (pain001.api.ui) and the application module can both depend on it without importing each other.

pain001.api.auth.API_KEY_ENV = 'PAIN001_API_KEY'

Environment variable that, when set, turns bearer-token auth on.

pain001.api.auth.require_api_key(authorization: str | None = Header(None)) → None

Enforce bearer-token auth when PAIN001_API_KEY is configured.

When the environment variable is unset the API remains open (local development mode); set it in any shared deployment.

Parameters:

authorization – The raw Authorization header, if any.

Raises:

HTTPException – 401 when a key is configured and the header does not carry it.

pain001.api.guards module

Taint barriers shared by the REST routers.

Pydantic already constrains request fields at the model boundary, but static analysis does not track enum membership. Every router that turns a request value into a filesystem path routes it through one of these helpers first: an explicit allow-list check the CodeQL py/path-injection query recognises as a sanitiser, registered as a neutral summary in .github/codeql/extensions/pain001-security.model.yml.

pain001.api.guards.sanitise_message_type(message_type: str) → str

Re-validate message_type against the fixed allow-list.

Parameters:

message_type – The string value of a request’s message-type enum.

Returns:

The canonical constant for that message type, guaranteed to be one of pain001.constants.valid_xml_types and therefore safe to join into the bundled template and schema paths.

Raises:

HTTPException – 400 if the value is not in the allow-list.

pain001.api.job_manager module

Job management for async XML generation.

class pain001.api.job_manager.JobManager(max_jobs: int = 1000, store: JobStore | None = None)

Bases: object

Manages async job lifecycle.

Parameters:

• max_jobs – Maximum number of jobs to keep in memory.

• store – Optional persistence backend. When provided, jobs are write-through persisted and rehydrated from it on construction so they survive a process restart.

cancel_job(job_id: str) → bool

Cancel a job.

Parameters:

job_id – Job identifier.

Returns:

True if cancelled, False if not found.

cleanup_old_jobs(keep_count: int = 100) → None

Remove old completed jobs to free memory.

Parameters:

keep_count – Number of recent jobs to keep.

create_job() → str

Create a new job.

Returns:

Job ID.

get_job(job_id: str) → JobResult | None

Get job by ID.

Parameters:

job_id – Job identifier.

Returns:

JobResult or None if not found.

update_status(job_id: str, status: JobStatus, progress: int = 0, result: dict[str, Any] | None = None, error: str | None = None) → None

Update job status.

Parameters:

• job_id – Job identifier.

• status – New status.

• progress – Progress percentage (0-100).

• result – Result data if completed.

• error – Error message if failed.

Terminal statuses (success/failed/cancelled) are final: a late-arriving update (e.g. a worker finishing after the user cancelled the job) must not resurrect or overwrite them.

class pain001.api.job_manager.JobResult(job_id: str, status: JobStatus, result: dict[str, Any] | None = None, error: str | None = None)

Bases: object

Represents a job result.

Parameters:

• job_id – Unique job identifier.

• status – Current job status.

• result – Job result data.

• error – Error message if failed.

classmethod from_dict(snapshot: dict[str, Any]) → JobResult

Reconstruct a job from a persisted snapshot.

Parameters:

snapshot – A dictionary previously produced by to_dict().

Returns:

The rehydrated JobResult.

to_dict() → dict[str, Any]

Serialise the job to a plain dictionary.

Returns:

A JSON-serialisable snapshot of the job’s state.

class pain001.api.job_manager.JobStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Job status enumeration.

CANCELLED = 'cancelled'

FAILED = 'failed'

PENDING = 'pending'

PROCESSING = 'processing'

SUCCESS = 'success'

pain001.api.job_store module

Pluggable persistence for async generation jobs.

By default the JobManager keeps jobs in memory, which is lost on restart. Setting PAIN001_JOB_STORE_DIR to a writable directory activates the FileJobStore, which write-through persists each job as a JSON document and rehydrates them on startup — so a job submitted before a deploy can still be polled afterwards.

class pain001.api.job_store.FileJobStore(directory: str | PathLike[str])

Bases: object

A JobStore that persists each job as a JSON file.

Parameters:

directory – Directory under which <job_id>.json files are kept. It is created if it does not exist.

delete(job_id: str) → None

Delete a persisted job snapshot if present.

Parameters:

job_id – Unique job identifier.

load_all() → dict[str, dict[str, Any]]

Load all persisted job snapshots, skipping unreadable files.

Returns:

A mapping of job id to its snapshot.

save(job_id: str, snapshot: dict[str, Any]) → None

Atomically persist a job snapshot as JSON.

Parameters:

• job_id – Unique job identifier.

• snapshot – Serialisable job state.

class pain001.api.job_store.JobStore(*args, **kwargs)

Bases: Protocol

Persistence backend for job snapshots.

delete(job_id: str) → None

Remove a persisted job snapshot.

Parameters:

job_id – Unique job identifier.

load_all() → dict[str, dict[str, Any]]

Load every persisted job snapshot.

Returns:

A mapping of job id to its snapshot.

save(job_id: str, snapshot: dict[str, Any]) → None

Persist a single job snapshot.

Parameters:

• job_id – Unique job identifier.

• snapshot – Serialisable job state.

class pain001.api.job_store.RedisJobStore(url: str | None = None, namespace: str = 'pain001:jobs', client: Any = None)

Bases: object

A JobStore that persists each job under a Redis hash key.

Multiple API replicas behind a load balancer share one Redis, so a job submitted on replica A can be polled on replica B and survives rolling restarts on either side. Job snapshots are serialised to JSON and stored at <namespace>:<job_id>; load_all enumerates every key under the namespace via SCAN (cursor-paginated, so it is safe on a large keyspace).

Parameters:

• url – Redis connection URL (redis://host:port/db).

• namespace – Optional key prefix; defaults to pain001:jobs. Use a per-environment value to isolate dev / staging / prod in the same instance.

• client – Optional pre-built redis.Redis instance; takes precedence over url (useful for tests with fakeredis).

Raises:

• ImportError – If the redis package is not installed.

• ValueError – If neither a url nor a client is supplied.

delete(job_id: str) → None

Remove a persisted job snapshot.

Parameters:

job_id – Unique job identifier.

load_all() → dict[str, dict[str, Any]]

Load every persisted job snapshot.

Returns:

A mapping of job id to its snapshot.

save(job_id: str, snapshot: dict[str, Any]) → None

Persist a single job snapshot.

Parameters:

• job_id – Unique job identifier.

• snapshot – Serialisable job state.

pain001.api.job_store.job_store_from_env() → JobStore | None

Build a job store from environment configuration.

Selection order:

• PAIN001_JOB_STORE_URL=redis://... -> RedisJobStore (multi-replica durable, requires the pain001[redis] extra).

• PAIN001_JOB_STORE_DIR=/path -> FileJobStore (single-replica durable, no extra dependency).

• Neither set -> None (the JobManager stays in-memory).

Returns:

A configured store, or None when persistence is not enabled.

Raises:

ValueError – If PAIN001_JOB_STORE_URL has an unrecognised scheme.

pain001.api.metrics module

Dependency-free Prometheus metrics for the Pain001 REST API.

A tiny in-process counter registry plus an ASGI middleware that counts HTTP requests by method and status. render_prometheus() renders the counters together with build info and live gauges (supported message types, scheme profiles, and async jobs by status) in the Prometheus text exposition format, served at GET /metrics.

This is a single-process exporter — sufficient for one API node. Scraped behind a gateway or aggregated with a push-gateway for multi-node setups.

class pain001.api.metrics.MetricsMiddleware(app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]])

Bases: object

ASGI middleware counting HTTP requests by method and status.

Parameters:

app – The wrapped ASGI application.

class pain001.api.metrics.MetricsRegistry

Bases: object

A minimal counter and summary registry rendering Prometheus text format.

inc(name: str, labels: dict[str, str] | None = None, value: float = 1.0) → None

Increment a counter.

Parameters:

• name – Metric name (Prometheus convention, e.g. foo_total).

• labels – Optional label set.

• value – Amount to add (default 1).

observe(name: str, value: float, labels: dict[str, str] | None = None) → None

Observe a value in a summary metric.

Parameters:

• name – Summary metric name (e.g. pain001_processing_seconds).

• value – Numeric observation to record.

• labels – Optional label set.

record_data_volumes(data: list[dict[str, Any]]) → None

Extract and record payment volumes in cents by currency from rows.

Parameters:

data – List of payment record dictionaries.

record_file_processed(status: str, message_type: str) → None

Record a file processing completion counter.

Parameters:

• status – Processing status (‘success’ or ‘failure’).

• message_type – ISO 20022 message definition identifier.

record_payment_volume(currency: str, cents: int | float) → None

Record payment amount volume in cents.

Parameters:

• currency – Three-letter ISO 4217 currency code.

• cents – Amount in smallest currency units (cents).

record_processing_seconds(seconds: float, labels: dict[str, str] | None = None) → None

Record processing duration in seconds into summary.

Parameters:

• seconds – Processing time elapsed in seconds.

• labels – Optional label set.

render(metric_name: str | None = None) → list[str]

Render the counters as Prometheus exposition lines.

Parameters:

metric_name – Optional metric name filter.

Returns:

One text line per counter sample.

render_summaries(metric_name: str | None = None, quantiles: tuple[float, ...] = (0.5, 0.95, 0.99)) → list[str]

Render summaries with quantiles, sum, and count samples.

Parameters:

• metric_name – Optional metric name filter.

• quantiles – Quantiles to calculate and render.

Returns:

Text lines for the summary quantiles, sum, and count.

reset() → None

Clear all counters and summaries (used by tests).

pain001.api.metrics.render_prometheus(version: str) → str

Render the full metrics document in Prometheus text format.

Parameters:

version – The running package version (for pain001_build_info).

Returns:

The complete exposition body (ending with a trailing newline).

pain001.api.models module

Pydantic models for FastAPI request/response validation.

class pain001.api.models.DataSourceType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Supported data source types.

CSV = 'csv'

JSON = 'json'

JSONL = 'jsonl'

PARQUET = 'parquet'

SQLITE = 'sqlite'

class pain001.api.models.GenerateXMLRequest(*, rules: Annotated[str | None, MinLen(min_length=1), MaxLen(max_length=65536)] = None, data_source: DataSourceType, file_path: str, message_type: MessageType = MessageType.PAIN_001_03, output_dir: str | None = None, validate_only: bool = False, table_name: str | None = None, scheme: str | None = None)

Bases: BaseModel

Request model for XML generation.

data_source: DataSourceType

file_path: str

message_type: MessageType

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'data_source': 'csv', 'file_path': 'payments.csv', 'message_type': 'pain.001.001.03', 'output_dir': '.', 'validate_only': False}]}, 'use_enum_values': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

output_dir: str | None

rules: str | None

scheme: str | None

table_name: str | None

validate_only: bool

class pain001.api.models.GenerateXMLResponse(*, success: bool, message: str, file_path: str | None = None, validation_errors: list[~pain001.api.models.ValidationError] = <factory>, scheme_violations: list[dict[str, ~typing.Any]] = <factory>)

Bases: BaseModel

Response model for XML generation.

file_path: str | None

message: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scheme_violations: list[dict[str, Any]]

success: bool

validation_errors: list[ValidationError]

class pain001.api.models.HealthResponse(*, status: str, version: str, message: str)

Bases: BaseModel

Response model for health check.

message: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: str

version: str

class pain001.api.models.JobStatusResponse(*, job_id: str, status: str, message: str, result: GenerateXMLResponse | None = None, error: str | None = None, progress_percent: int = 0)

Bases: BaseModel

Response model for job status.

error: str | None

job_id: str

message: str

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

progress_percent: int

result: GenerateXMLResponse | None

status: str

class pain001.api.models.MessageType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Supported ISO 20022 message types.

PAIN_001_03 = 'pain.001.001.03'

PAIN_001_04 = 'pain.001.001.04'

PAIN_001_05 = 'pain.001.001.05'

PAIN_001_06 = 'pain.001.001.06'

PAIN_001_07 = 'pain.001.001.07'

PAIN_001_08 = 'pain.001.001.08'

PAIN_001_09 = 'pain.001.001.09'

PAIN_001_10 = 'pain.001.001.10'

PAIN_001_11 = 'pain.001.001.11'

PAIN_001_12 = 'pain.001.001.12'

PAIN_001_13 = 'pain.001.001.13'

PAIN_008_02 = 'pain.008.001.02'

PAIN_008_08 = 'pain.008.001.08'

class pain001.api.models.UiFilePayload(*, filename: Annotated[str, MinLen(min_length=1), MaxLen(max_length=255)], content: str, message_type: MessageType = MessageType.PAIN_001_03, scheme: str | None = None)

Bases: BaseModel

An uploaded payment file, as the dashboard sends it.

The browser reads the dropped file and posts its text; the server parses it in memory, so nothing the user uploads touches the disk.

content: str

filename: str

message_type: MessageType

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'content': 'id,date,nb_of_txs,...', 'filename': 'payments.csv', 'message_type': 'pain.001.001.03', 'scheme': 'sepa-sct,anti-duplicate'}]}, 'use_enum_values': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scheme: str | None

class pain001.api.models.UiGenerateResponse(*, success: bool, message: str, filename: str | None = None, xml: str | None = None, validation_errors: list[~pain001.api.models.ValidationError] = <factory>, scheme_violations: list[dict[str, ~typing.Any]] = <factory>)

Bases: BaseModel

Result of generating XML from an uploaded file.

On success the XML document travels inline so the browser can offer it as a download; on failure the blocking findings are attached.

filename: str | None

message: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scheme_violations: list[dict[str, Any]]

success: bool

validation_errors: list[ValidationError]

xml: str | None

class pain001.api.models.ValidationError(*, field: str, message: str, value: Any | None = None)

Bases: BaseModel

Validation error details.

field: str

message: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

value: Any | None

class pain001.api.models.ValidationRequest(*, rules: Annotated[str | None, MinLen(min_length=1), MaxLen(max_length=65536)] = None, data_source: DataSourceType, file_path: str, message_type: MessageType = MessageType.PAIN_001_03, table_name: str | None = None, scheme: str | None = None)

Bases: BaseModel

Request model for data validation.

data_source: DataSourceType

file_path: str

message_type: MessageType

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'data_source': 'csv', 'file_path': 'payments.csv', 'message_type': 'pain.001.001.03', 'scheme': 'sepa-sct'}]}, 'use_enum_values': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rules: str | None

scheme: str | None

table_name: str | None

class pain001.api.models.ValidationResponse(*, is_valid: bool, total_rows: int, valid_rows: int = 0, invalid_rows: int = 0, errors: list[~pain001.api.models.ValidationError] = <factory>, scheme_violations: list[dict[str, ~typing.Any]] = <factory>)

Bases: BaseModel

Response model for validation results.

classmethod calculate_invalid_rows(v: int, info: ValidationInfo) → int

Calculate invalid rows from total and valid counts.

Parameters:

• v – Current invalid_rows value.

• info – Validation info containing all field values.

Returns:

Calculated invalid rows (total - valid).

errors: list[ValidationError]

invalid_rows: int

is_valid: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scheme_violations: list[dict[str, Any]]

total_rows: int

valid_rows: int

pain001.api.ratelimit module

In-process, dependency-free rate limiting for the Pain001 REST API.

A small fixed-window limiter keyed by client IP, configured entirely through the PAIN001_RATE_LIMIT environment variable (e.g. 100/minute or 20/second). When the variable is unset the middleware is a no-op, so local development is unaffected.

This is deliberately a single-process limiter — sufficient for one API node and for protecting against accidental floods. Horizontally-scaled deployments should enforce limits at the gateway or with a shared store (e.g. Redis); see the API documentation.

class pain001.api.ratelimit.InProcessFixedWindowBackend(max_requests: int, window_seconds: float, clock: Callable[[], float] | None = None)

Bases: object

Single-process fixed-window backend (the default).

Parameters:

• max_requests – Maximum requests per window.

• window_seconds – Length of the window.

• clock – Monotonic time source (injectable for tests).

is_allowed(key: str) → bool

Record a hit and return whether the request stays within the cap.

class pain001.api.ratelimit.RateLimitMiddleware(app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], max_requests: int, window_seconds: float, clock: Callable[[], float] | None = None, backend: RateLimiterBackend | None = None)

Bases: object

ASGI middleware enforcing a fixed-window per-client request cap.

Accepts a pluggable backend (see RateLimiterBackend). For backward compatibility, callers may pass max_requests + window_seconds directly; the middleware then builds an InProcessFixedWindowBackend itself. Either form works.

Parameters:

• app – The wrapped ASGI application.

• max_requests – Maximum requests allowed per window (ignored if backend is supplied).

• window_seconds – Length of the window in seconds (also used for the Retry-After header).

• clock – Monotonic time source (injectable for tests; only used when constructing a fallback in-process backend).

• backend – Pluggable backend; pass an instance to use Redis or a custom limiter.

class pain001.api.ratelimit.RateLimiterBackend(*args, **kwargs)

Bases: Protocol

A pluggable rate-limit decision backend.

is_allowed(key: str) → bool

Record a hit and return True when within the cap.

Parameters:

key – Per-client bucket key (e.g. client IP).

Returns:

True when the hit fits within the current window’s cap; False when the window is full and the request should be rejected with HTTP 429.

class pain001.api.ratelimit.RedisFixedWindowBackend(max_requests: int, window_seconds: float, url: str | None = None, namespace: str = 'pain001:ratelimit', clock: Callable[[], float] | None = None, client: Any = None)

Bases: object

Redis-backed fixed-window backend for multi-replica deployments.

Uses INCR + EXPIRE on a per-bucket key namespaced as <namespace>:<window-floor>:<client-key> so two API replicas sharing one Redis enforce the cap consistently. Bucket keys carry a TTL equal to the window length, so stale buckets don’t accumulate.

Parameters:

• max_requests – Maximum requests per window.

• window_seconds – Length of the window.

• url – Redis connection URL (redis://host:port/db).

• namespace – Optional key prefix; defaults to pain001:ratelimit.

• clock – Time source (injectable for tests; returns seconds since epoch, not monotonic, so window floors line up across hosts).

• client – Optional pre-built redis.Redis instance; takes precedence over url (useful for tests with fakeredis).

Raises:

• ImportError – If the redis package is not installed.

• ValueError – If neither a url nor a client is supplied.

is_allowed(key: str) → bool

Record a hit and return whether the request stays within the cap.

pain001.api.ratelimit.backend_from_env(max_requests: int, window_seconds: float, clock: Callable[[], float] | None = None) → RateLimiterBackend

Select a backend from environment configuration.

Selection order:

• PAIN001_RATE_LIMIT_BACKEND=redis (default URL from PAIN001_RATE_LIMIT_REDIS_URL or PAIN001_JOB_STORE_URL).

• Anything else (or unset) -> the in-process default.

Parameters:

• max_requests – Per-window cap.

• window_seconds – Window length.

• clock – Optional time source (passed through).

Returns:

A configured backend.

Raises:

ValueError – If PAIN001_RATE_LIMIT_BACKEND is set to an unrecognised value.

pain001.api.ratelimit.parse_rate_limit(spec: str) → tuple[int, float]

Parse a <count>/<window> rate-limit specification.

Parameters:

spec – A specification such as 100/minute or 5/second.

Returns:

A (max_requests, window_seconds) tuple.

Raises:

ValueError – If the specification is malformed or the window is not one of second/minute/hour/day.

pain001.api.ui module

Single-file hosted dashboard at /api/v1/ui (issue #188).

A finance user without Python or a terminal opens one URL, drops a CSV on the page, picks a message type and the scheme rulebooks to enforce, and downloads the validated XML. The page is one vanilla HTML document with inline CSS and JavaScript: no framework, no build step, no CDN, no new runtime dependency. It talks to two JSON endpoints that sit next to it:

• POST /api/v1/ui/validate - the file’s text plus options in, the same ValidationResponse the path-based /validate returns out.

• POST /api/v1/ui/generate - the same input; the XML document comes back inline so the browser can offer it as a download without the server ever writing the batch to disk.

Both endpoints carry the file content rather than a path because a browser cannot name a file on the server, which is what the path-based endpoints expect. They honour PAIN001_API_KEY like every other route; the page keeps the key in sessionStorage for the tab’s lifetime only. Operators running a pure-API deployment can switch the whole surface off with PAIN001_UI_DISABLED=1.

pain001.api.ui.MAX_UPLOAD_BYTES = 10485760

Largest file the dashboard accepts, in UTF-8 bytes (10 MiB).

pain001.api.ui.SUPPORTED_SUFFIXES: tuple[str, ...] = ('.csv', '.json', '.jsonl')

Text formats the dashboard parses in memory.

pain001.api.ui.UI_DISABLED_ENV = 'PAIN001_UI_DISABLED'

Set to 1 / true / yes to take the dashboard offline.

pain001.api.ui.is_ui_enabled() → bool

Return True unless the operator set PAIN001_UI_DISABLED.

Returns:

False when the env var is one of 1 / true / yes (case-insensitive); True otherwise.

async pain001.api.ui.ui_generate(payload: UiFilePayload) → UiGenerateResponse

Validate the upload, then render the XML and return it inline.

Parameters:

payload – The file’s text, its name, the message type, and an optional scheme spec.

Returns:

success=True with the XML document and a suggested filename, or success=False carrying the validation errors or scheme violations that blocked generation.

Raises:

HTTPException – 400 when rendering fails (for example the rendered document does not satisfy the XSD).

async pain001.api.ui.ui_page() → HTMLResponse

Serve the dashboard page.

Returns:

The single-file HTML application, with the supported message types and scheme profiles of this build embedded.

async pain001.api.ui.ui_validate(payload: UiFilePayload) → ValidationResponse

Validate the uploaded text against the schema and optional schemes.

Parameters:

payload – The file’s text, its name, the message type, and an optional scheme spec.

Returns:

The same shape as POST /validate.