Validation Engine¶

The standalone package includes a five-layer validation engine (craft_easy_file_import.validation). Layers run in a fixed order; later layers can assume earlier ones have already passed.

Layer 1 — Schema      (JSON Schema draft 2020-12, per row)
Layer 2 — Row         (domain-specific checks: IBAN, BIC, regex, etc.)
Layer 3 — Business    (read-only API lookups with caching)
Layer 4 — Plugin      (entry-point-registered Python callables)
Layer 5 — Cross-row   (two-pass: accumulate per row, validate after all rows)

Layers 1–4 run per row via ValidationEngine.validate_row(). Layer 5 runs once via ValidationEngine.finalize() after all rows have been processed.

All layers produce ValidationError objects — no layer raises exceptions in the happy path. The caller decides what to do based on severity:

Severity	Effect
`error`	Row becomes a `RejectedRecord`
`warning`	Row is committed; warning attached for later review
`info`	Logged; no impact on the row

Wiring the engine¶

The engine is configured through the validation dict on ImportTemplate:

from craft_easy_file_import.models import ImportTemplate
from craft_easy_file_import.validation.engine import ValidationEngine

template = ImportTemplate(
    id="my-template",
    validation={
        "schema": {...},          # Layer 1
        "row_rules": [...],       # Layer 2
        "business_rules": [...],  # Layer 3
        "plugins": [...],         # Layer 4
        "aggregates": [...],      # Layer 5 — aggregate definitions
        "cross_row_rules": [...], # Layer 5 — rules against aggregates
        "abort_on_first_error": False,  # stop per-row after first ERROR
        "locale": "sv",           # message translation locale
    },
)

engine = ValidationEngine(template, api_client=api)

for i, row in enumerate(rows, start=1):
    errors = await engine.validate_row(row, row_number=i)
    # handle errors...

cross_row_errors = await engine.finalize()

Layer 1: Schema (JSON Schema draft 2020-12)¶

Validates each row dict against a JSON Schema definition. Any keyword supported by jsonschema's Draft202012Validator works.

"schema": {
    "type": "object",
    "required": ["org_nr", "amount", "currency"],
    "properties": {
        "org_nr":   {"type": "string", "minLength": 10},
        "amount":   {"type": "number", "minimum": 0},
        "currency": {"type": "string", "enum": ["SEK", "EUR", "NOK"]},
        "due_date": {"type": "string", "format": "date"},
    },
    "additionalProperties": True,
}

Schema errors carry layer="schema", rule="json_schema.<keyword>", and a field derived from the JSON Pointer path to the failing node.

Layer 2: Row rules¶

Declarative row-level checks for domain formats that JSON Schema cannot express concisely. Rules are evaluated in order; all failures are collected unless abort_on_first_error is set.

Every rule has type, field, optional severity (error by default), and optional allow_empty (skip the rule when the field is empty).

regex¶

{"type": "regex", "field": "postal_code", "pattern": r"^\d{5}$"}

luhn¶

Validates a Luhn (mod-10) check digit — card numbers, some ID formats:

{"type": "luhn", "field": "card_number"}

iban¶

Full IBAN validation: country prefix check, length-by-country, and mod-97:

{"type": "iban", "field": "bank_account", "allow_empty": True}

Accepts IBANs with spaces or hyphens (SE35 5000 0000 0549 1000 0003 and SE3550000000054910000003 both pass).

bic¶

BIC/SWIFT format validation (8 or 11 characters):

{"type": "bic", "field": "bic_code"}

orgnr_se¶

Swedish organisation number (10 digits, third digit ≥ 2, Luhn check). Accepts 10- and 12-digit forms with or without hyphen:

{"type": "orgnr_se", "field": "org_nr"}

personnr_se¶

Swedish personal identity number (YYYYMMDD-XXXX or YYMMDD-XXXX, Luhn check on the 10-digit form):

{"type": "personnr_se", "field": "pnr", "allow_empty": True}

date_range¶

Validates that a date falls within a range. min/max accept ISO dates or the string "today". Parsed formats: YYYY-MM-DD, YYYYMMDD, DD/MM/YYYY, DD.MM.YYYY.

# Payments must not be in the future
{"type": "date_range", "field": "payment_date", "max": "today"}

# Invoice dates must be within the last 5 years
{"type": "date_range", "field": "invoice_date", "min": "2020-01-01"}

# Custom format
{"type": "date_range", "field": "val_dt", "format": "%Y%m%d", "min": "today"}

one_of¶

Exact membership check, with optional case-insensitive mode:

{"type": "one_of", "field": "status", "values": ["active", "suspended", "closed"]}
{"type": "one_of", "field": "country", "values": ["SE", "NO", "DK"], "case_sensitive": False}

Layer 3: Business rules¶

Business rules make read-only API calls to enforce referential integrity. Results are cached per import run in ValidationContext to avoid N+1 queries.

counterparty_exists¶

Checks that a field value exists in another collection:

{
    "type": "counterparty_exists",
    "field": "supplier_org_nr",
    "collection": "suppliers",
    "lookup_field": "org_nr",
    "severity": "error",
}

Calls api_client.exists("suppliers", {"org_nr": value}). Repeated lookups for the same value within a run hit the cache.

unique_field¶

Checks that a field value is unique across all rows in the current run:

{"type": "unique_field", "field": "reference_number", "severity": "error"}

Uniqueness is checked in-memory within the run — not across previous runs.

Layer 4: Plugins¶

Plugins are Python callables registered via pyproject.toml entry points. Templates reference them by name; arbitrary module paths are not allowed (security model matches mapper plugins).

Registering a plugin¶

# pyproject.toml
[project.entry-points."craft_easy_file_import.validators"]
swedish_parking_fee = "mypackage.import_validators:validate_parking_fee"

The callable signature:

from craft_easy_file_import.validation.context import ValidationContext
from craft_easy_file_import.validation.errors import ValidationError

async def validate_parking_fee(
    row: dict,
    context: ValidationContext,
    api_client: Any,
) -> list[ValidationError]:
    # return [] for pass, or [ValidationError(...)] for failures
    ...

Sync functions are also accepted — the engine detects and awaits accordingly.

Using a plugin in a template¶

"plugins": ["swedish_parking_fee"]

Plugin load failures (import errors) are logged at WARNING level but do not prevent other plugins from running. A plugin that raises at runtime produces a single ValidationError with severity=error and rule="plugin.{name}.error".

Programmatic registration (testing)¶

from craft_easy_file_import.validation.plugin import register_plugin

register_plugin("my_validator", my_validate_fn)

Layer 5: Cross-row validation¶

Two-pass validation that computes aggregates across all rows, then runs rules against those aggregates.

Pass 1 (accumulate) runs during row iteration and updates running aggregates. Pass 2 (finalize) runs once after all rows and checks the rules.

Aggregates¶

Each aggregate has a name, an op, and (for field-based ops) a field:

`op`	What it computes
`sum`	Running sum of the field (numeric coercion)
`count`	Total number of rows
`distinct`	Unique values of the field (stored as a set; resolved to count in rules)
`min`	Minimum value of the field
`max`	Maximum value of the field

"aggregates": [
    {"name": "total_amount",    "op": "sum",      "field": "amount"},
    {"name": "row_count",       "op": "count"},
    {"name": "unique_orgnrs",   "op": "distinct", "field": "org_nr"},
]

Cross-row rules¶

Rules compare two expressions using a comparison operator. Expressions can reference aggregates (aggregates.name), file header fields (header.name), or literals.

`type`	Check
`equals`	left == right
`less_than`	left < right
`less_than_or_equal`	left <= right
`greater_than`	left > right
`greater_than_or_equal`	left >= right

"cross_row_rules": [
    # Total amount in the file must match a control total in the file header
    {
        "type": "equals",
        "left": "aggregates.total_amount",
        "right": "header.control_total",
        "severity": "error",
    },
    # There must be at least 1 row
    {
        "type": "greater_than_or_equal",
        "left": "aggregates.row_count",
        "right": "1",
    },
    # At most 500 unique org numbers
    {
        "type": "less_than_or_equal",
        "left": "aggregates.unique_orgnrs",
        "right": "500",
        "severity": "warning",
    },
]

The header dict is passed to finalize() — typically populated from file-level metadata by the parser or pipeline:

cross_row_errors = await engine.finalize(header={"control_total": "12500.00"})

Complete example¶

template = ImportTemplate(
    id="supplier-payments",
    validation={
        "schema": {
            "type": "object",
            "required": ["org_nr", "amount", "due_date"],
            "properties": {
                "org_nr":   {"type": "string"},
                "amount":   {"type": "number"},
                "due_date": {"type": "string"},
            },
        },
        "row_rules": [
            {"type": "orgnr_se",   "field": "org_nr"},
            {"type": "date_range", "field": "due_date", "min": "today"},
        ],
        "business_rules": [
            {
                "type": "counterparty_exists",
                "field": "org_nr",
                "collection": "suppliers",
                "lookup_field": "org_nr",
            },
        ],
        "aggregates": [
            {"name": "total",     "op": "sum",   "field": "amount"},
            {"name": "row_count", "op": "count"},
        ],
        "cross_row_rules": [
            {
                "type": "greater_than",
                "left": "aggregates.row_count",
                "right": "0",
            },
        ],
    },
)