Validation & business-rule pipeline

Python SDK

aiotic.pipeline is the integration layer's answer to ERPs that cannot validate an order themselves. It runs three kinds of steps in order and produces a Verdict:

ErpPurchaseOrder ──▶ Sanitizers ──▶ Validators ──▶ Business rules ──▶ Verdict(ok | errors, warnings, cleaned order)

Sanitizer — apply(order, ctx) -> order: returns a cleaned copy.
Validator — check(order, ctx) -> Iterable[Issue]: data shape and consistency.
BusinessRule — same contract; policy. Kept in a separate list for readability.

An Issue has a code, a human message, a severity (ERROR rejects, WARNING is recorded) and an optional path (items[2].article_number).

python

from aiotic.pipeline import Pipeline, sanitizers as S, validators as V, rules as R

pipeline = Pipeline(
    sanitizers=[S.StripWhitespace(), S.NormalizeCountryCodes(), S.NormalizeCurrency(), S.NormalizePostalCodes(), S.MapUnits(), S.FillShippingFromCustomer(), S.DropEmptyLines()],
    validators=[V.RequiredFields(), V.CustomerResolved(erp), V.ArticlesInCatalog(erp), V.PositiveQuantities(), V.LineTotalsConsistent(), V.OrderTotalConsistent(), V.CurrencyAllowed(["EUR"]), V.DeliveryDateSane()],
    rules=[R.NoDuplicateOrder(store), R.CustomerNotBlocked(erp), R.OrderValueWithin(max_total=50_000), R.ShipToAddressComplete()],
)

verdict = pipeline.run(order, {"request_id": request_id})
if not verdict.ok:
    return ErpReceiveResponse.rejected(verdict.error_message())     # shown to the operator
erp.create_sales_order(request_id, verdict.order)                  # the cleaned order

aiotic.service.default_pipeline(erp, store) builds this configuration (adapting to what the adapter implements).

Built-in sanitizers

Class	Effect
`StripWhitespace()`	trim + collapse whitespace on every string; empty → `null`
`NormalizeCountryCodes(extra=None, default=None)`	`Nederland`/`NLD` → `NL`; extend with your own aliases
`NormalizeCurrency(default="EUR")`	`€` → `EUR`, uppercase; fills missing line currencies from the header
`NormalizePostalCodes()`	`7327aa` → `7327 AA` (NL); trims others
`NormalizeOrderNumber(pattern, replacement="-", max_length=35)`	strip characters your reference field cannot hold; truncate
`MapUnits(mapping=None, default_unit="ST", keep_unknown=True)`	`stuks`/`Stk`/`pcs` → `ST`, `mtr` → `M`, … plus your map
`DropEmptyLines()`	remove lines with neither quantity, article nor price
`FillShippingFromCustomer()`	empty ship-to → copy of the customer block
`Custom(name, fn)`	any `(order, ctx) -> order`

Built-in validators

Class	Rejects when
`RequiredFields(require_customer_id=True, require_delivery_date=False)`	no order number / date / lines / customer id
`CustomerResolved(customers: CustomerPort)`	`customer_id` unknown in the ERP (stores the ERP record in `ctx["erp_customer"]`)
`ArticlesInCatalog(catalog: CatalogPort)`	a line has no article or an unknown one
`PositiveQuantities(max_quantity=100000)`	quantity missing / ≤ 0; warns above the max
`LineTotalsConsistent(tolerance=0.05, severity=WARNING)`	`quantity × price ≠ line_total`
`OrderTotalConsistent(tolerance_ratio=0.25, severity=WARNING)`	lines sum far from the document total
`CurrencyAllowed(["EUR"])`	header or line currency not in the set
`DeliveryDateSane(max_days_in_past=30, max_days_ahead=365)`	not ISO; warns when implausible
`Custom(name, fn)`	any `(order, ctx) -> Iterable[Issue]`

Built-in business rules

Class	Rejects when
`NoDuplicateOrder(store)`	same `(customer_id, order_number)` was booked before (remembered only after a successful booking)
`CustomerNotBlocked(customers: CustomerPort)`	`is_blocked(customer_id)`
`OrderValueWithin(min_total=None, max_total=None, severity=WARNING)`	value outside the band
`ShipToAddressComplete()`	street / postal code / city missing
`@rule("name")` decorator	turn a generator function into a rule

Writing your own

python

from aiotic.pipeline import Issue, Severity
from aiotic.pipeline.rules import rule

@rule("credit_limit")
def credit_limit(order, ctx):
    cust = ctx.get("erp_customer")                  # set by CustomerResolved
    total = order.total_price or sum(i.line_total or 0 for i in order.items)
    if cust and cust.get("credit_limit") is not None and total > cust["credit_limit"]:
        yield Issue("credit_limit", f"Order value {total:.2f} exceeds credit limit {cust['credit_limit']:.2f}",
                    path="total_price", severity=Severity.ERROR)

Guidelines:

Yield Issues; do not raise. One issue per problem, with a path.
Messages are read by operators — say what is wrong and where.
Keep steps pure; the pipeline deep-copies the order once and hands each step the current version.
Use ctx to pass lookups between steps instead of querying twice.
Set stop_on_first_error=True on the pipeline if later checks are expensive and pointless after a rejection.

Verdict

Member	Meaning
`ok`	no `ERROR` issues
`errors`, `warnings`	filtered issue lists
`order`	the sanitized order — book this, not the input
`steps`	which steps ran (for logs)
`error_message(limit=5)`	one line for the operator

Warnings are worth logging and, for model C automation, worth turning into "needs a human" even though the order is technically bookable.

Validation & business-rule pipeline ​

Built-in sanitizers ​

Built-in validators ​

Built-in business rules ​

Writing your own ​

Verdict ​