Validation when your ERP has no functional API

Receiving orders

Two kinds of ERP show up in integrations:

	Functional API	Data API / direct database
How you create an order	POST /salesOrders or a "create order" function	INSERT into order tables (or a generic "data" API that does the same)
Who validates	the ERP: unknown item, blocked customer, credit limit, unit conversion…	nobody — whatever you write lands; the ERP trips over it later
Failure mode	immediate, explicit error you can pass to the operator	silent corruption discovered days later by someone else

With a data API, the integration layer has to be the validation layer. That is what the SDK's pipeline is for.

Rendering diagram…

Sanitizers → validators → business rules → adapter. With a data-API ERP the pipeline is not optional.

What to validate, minimum set

Check	Why	SDK step
Article exists and is not blocked	the #1 way to create unbookable orders	validators.ArticlesInCatalog(erp)
Customer exists	orphan orders	validators.CustomerResolved(erp)
Customer not blocked / credit hold	policy	rules.CustomerNotBlocked(erp)
Quantities are positive integers	AIOTIC sends null for unreadable quantities on override sends	validators.PositiveQuantities()
Unit of measure known	stuks vs ST vs PCS	sanitizers.MapUnits({...}) (+ reject unknown)
Currency allowed	wrong currency = wrong price	validators.CurrencyAllowed(["EUR"])
Delivery date plausible	typo detection	validators.DeliveryDateSane()
Duplicate PO for the same customer	double bookings	rules.NoDuplicateOrder(store)
Ship-to address complete	undeliverable orders	rules.ShipToAddressComplete()
Line total vs quantity × price	mis-read quantities (18.000 vs 18)	validators.LineTotalsConsistent() (warning)

Putting it together

from aiotic.erp.data_api import DataApiAdapter
from aiotic.pipeline import Pipeline, sanitizers as S, validators as V, rules as R
from aiotic.receive import ErpReceiver, SqliteStore
import pyodbc

erp = DataApiAdapter(lambda: pyodbc.connect(DSN), paramstyle="qmark")   # implements ErpPort + CatalogPort + CustomerPort
store = SqliteStore("integration.db")

pipeline = Pipeline(
    sanitizers=[S.StripWhitespace(), S.NormalizeCountryCodes(), S.NormalizeCurrency(), S.MapUnits({"doos": "BOX"}), S.FillShippingFromCustomer()],
    validators=[V.RequiredFields(), V.CustomerResolved(erp), V.ArticlesInCatalog(erp), V.PositiveQuantities(), V.CurrencyAllowed(["EUR"])],
    rules=[R.CustomerNotBlocked(erp), R.NoDuplicateOrder(store), R.ShipToAddressComplete()],
)
receiver = ErpReceiver(erp, api_key=RECEIVE_KEY, pipeline=pipeline, store=store)

A rejecting verdict becomes {"success": false, "error": "Unknown article number: PROD-999 (items[2].article_number); Customer 10577 is blocked for new orders"} — the operator reads it in the AIOTIC app, fixes the order (or the master data) and sends again.

Your own rules

Anything that would live in an ERP's "before insert" logic can be a rule:

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

@rule("no_weekend_delivery")
def no_weekend(order, ctx):
    from datetime import date
    if order.delivery_date and date.fromisoformat(order.delivery_date).weekday() >= 5:
        yield Issue("weekend_delivery", "Delivery date falls on a weekend", path="delivery_date")

pipeline.rules.append(no_weekend)

ctx carries the request_id and anything earlier steps stored (for example the ERP customer record from CustomerResolved), so rules can use ERP data without a second lookup.

Sanitize before you validate

Order matters. Sanitizers normalise what the document printed (Nederland → NL, € → EUR, 7327aa → 7327 AA, stuks → ST); validators then judge the cleaned order. Everything returns a new order — the payload AIOTIC sent is never mutated, so you can log both.

Where the same checks live with a functional API

Keep the pipeline anyway, lighter: sanitizers plus RequiredFields and NoDuplicateOrder. Let the ERP do catalog and customer checks and pass its error text through (ErpRejected). You get one code path for both ERP styles, which is the point.