# Copyright (C) 2023-2026 Pain001. All rights reserved.
# SPDX-License-Identifier: Apache-2.0 OR MIT
#
# Licensed under either of the Apache License, Version 2.0 or the MIT
# License, at your option. You may not use this file except in
# compliance with one of those licences. Copies are provided in
# LICENSE-APACHE and LICENSE-MIT.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the Licences is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the applicable Licence for the specific language
# governing permissions and limitations.
"""Pydantic models for FastAPI request/response validation."""
# pylint: disable=too-few-public-methods
from enum import Enum
from typing import Any
from pydantic import (
BaseModel,
ConfigDict,
Field,
ValidationInfo,
field_validator,
)
[docs]
class DataSourceType(str, Enum):
"""Supported data source types."""
CSV = "csv"
SQLITE = "sqlite"
JSON = "json"
JSONL = "jsonl"
PARQUET = "parquet"
[docs]
class MessageType(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"
[docs]
class ValidationRequest(BaseModel): # pylint: disable=too-few-public-methods
"""Request model for data validation."""
rules: str | None = Field(
default=None,
min_length=1,
max_length=65536,
description="Inline YAML CEL policy; requires pain001[rules]. Never a server-side file path.",
)
data_source: DataSourceType = Field(
..., description="Type of data source (csv, sqlite, json, etc.)"
)
file_path: str = Field(..., description="Path to the data file")
message_type: MessageType = Field(
default=MessageType.PAIN_001_03,
description="ISO 20022 message type",
)
table_name: str | None = Field(
default=None,
description="Table name for SQLite sources",
)
scheme: str | None = Field(
default=None,
description="Payment-scheme rulebook to validate against "
"(e.g. 'sepa-sct', 'sepa-sdd', 'sepa-b2b', 'sepa-inst', 'xborder-ct', "
"'anti-duplicate'); comma-separate names to run several",
)
model_config = ConfigDict(
use_enum_values=False,
json_schema_extra={
"examples": [
{
"data_source": "csv",
"file_path": "payments.csv",
"message_type": "pain.001.001.03",
"scheme": "sepa-sct",
}
]
},
)
[docs]
class GenerateXMLRequest(BaseModel):
"""Request model for XML generation."""
rules: str | None = Field(
default=None,
min_length=1,
max_length=65536,
description="Inline YAML CEL policy enforced before generation; requires pain001[rules].",
)
data_source: DataSourceType = Field(..., description="Type of data source")
file_path: str = Field(..., description="Path to the data file")
message_type: MessageType = Field(
default=MessageType.PAIN_001_03,
description="ISO 20022 message type",
)
output_dir: str | None = Field(
default=None,
description="Output directory for generated XML",
)
validate_only: bool = Field(
default=False,
description="Only validate, don't generate XML",
)
table_name: str | None = Field(
default=None,
description="Table name for SQLite sources",
)
scheme: str | None = Field(
default=None,
description="Payment-scheme rulebook to enforce before generating "
"(e.g. 'sepa-sct', 'sepa-sdd', 'sepa-b2b', 'sepa-inst', 'xborder-ct', "
"'anti-duplicate'); comma-separate names to run several",
)
model_config = ConfigDict(
use_enum_values=False,
json_schema_extra={
"examples": [
{
"data_source": "csv",
"file_path": "payments.csv",
"message_type": "pain.001.001.03",
"output_dir": ".",
"validate_only": False,
}
]
},
)
[docs]
class ValidationError(BaseModel):
"""Validation error details."""
field: str = Field(..., description="Field name or JSON path")
message: str = Field(..., description="Error message")
value: Any | None = Field(None, description="The invalid value")
[docs]
class ValidationResponse(BaseModel):
"""Response model for validation results."""
is_valid: bool = Field(..., description="Whether data is valid")
total_rows: int = Field(..., description="Total number of rows")
valid_rows: int = Field(default=0, description="Number of valid rows")
invalid_rows: int = Field(default=0, description="Number of invalid rows")
errors: list[ValidationError] = Field(
default_factory=list,
description="List of validation errors",
)
scheme_violations: list[dict[str, Any]] = Field(
default_factory=list,
description="Scheme-rulebook violations (when a scheme is given)",
)
[docs]
@field_validator("invalid_rows", mode="after")
@classmethod
def calculate_invalid_rows(cls, v: int, info: ValidationInfo) -> int:
"""Calculate invalid rows from total and valid counts.
Args:
v: Current invalid_rows value.
info: Validation info containing all field values.
Returns:
Calculated invalid rows (total - valid).
"""
data = info.data
if "total_rows" in data and "valid_rows" in data:
return int(data["total_rows"]) - int(data["valid_rows"])
# total_rows failed its own validation; pydantic reports that error
return v
[docs]
class GenerateXMLResponse(BaseModel):
"""Response model for XML generation."""
success: bool = Field(..., description="Whether generation succeeded")
message: str = Field(..., description="Result message")
file_path: str | None = Field(None, description="Path to generated XML")
validation_errors: list[ValidationError] = Field(
default_factory=list,
description="Validation errors if validation failed",
)
scheme_violations: list[dict[str, Any]] = Field(
default_factory=list,
description="Scheme-rulebook violations if scheme validation failed",
)
[docs]
class JobStatusResponse(BaseModel):
"""Response model for job status."""
job_id: str = Field(..., description="Unique job identifier")
status: str = Field(
...,
description="Current job status (pending, processing, success, failed, cancelled)",
)
message: str = Field(..., description="Status message")
result: GenerateXMLResponse | None = Field(
None, description="Result when status is success"
)
error: str | None = Field(None, description="Error message if failed")
progress_percent: int = Field(
default=0, description="Progress percentage (0-100)"
)
model_config = ConfigDict(use_enum_values=True)
[docs]
class UiFilePayload(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.
"""
filename: str = Field(
...,
min_length=1,
max_length=255,
description="Name the browser reported; only the suffix "
"(.csv, .json, .jsonl) is used to pick a parser",
)
content: str = Field(..., description="The file's text (UTF-8)")
message_type: MessageType = Field(
default=MessageType.PAIN_001_03,
description="ISO 20022 message type",
)
scheme: str | None = Field(
default=None,
description="Payment-scheme rulebook(s) to enforce "
"(e.g. 'sepa-sct', 'anti-duplicate'); comma-separate names to "
"run several",
)
model_config = ConfigDict(
use_enum_values=False,
json_schema_extra={
"examples": [
{
"filename": "payments.csv",
"content": "id,date,nb_of_txs,...",
"message_type": "pain.001.001.03",
"scheme": "sepa-sct,anti-duplicate",
}
]
},
)
[docs]
class UiGenerateResponse(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.
"""
success: bool = Field(..., description="Whether generation succeeded")
message: str = Field(..., description="Result message")
filename: str | None = Field(
default=None, description="Suggested download name for the XML"
)
xml: str | None = Field(
default=None, description="The generated XML document"
)
validation_errors: list[ValidationError] = Field(
default_factory=list,
description="Validation errors if validation failed",
)
scheme_violations: list[dict[str, Any]] = Field(
default_factory=list,
description="Scheme-rulebook violations if scheme validation failed",
)
[docs]
class HealthResponse(BaseModel):
"""Response model for health check."""
status: str = Field(..., description="Service status")
version: str = Field(..., description="API version")
message: str = Field(..., description="Health check message")