ERP adapters (functional vs data API)

Python SDK

The receive endpoint talks to your ERP through one small interface. Implement it once; everything else (key check, idempotency, pipeline, response) is shared.

The ports

class ErpPort(Protocol):
    def find_order_by_request_id(self, request_id: str) -> ErpCreateResult | None: ...   # idempotency
    def create_sales_order(self, request_id: str, order: ErpPurchaseOrder) -> ErpCreateResult: ...

class CatalogPort(Protocol):                      # used by validators.ArticlesInCatalog
    def product_exists(self, article_number: str) -> bool: ...

class CustomerPort(Protocol):                     # used by validators.CustomerResolved, rules.CustomerNotBlocked
    def get_customer(self, customer_id: str) -> dict | None: ...
    def is_blocked(self, customer_id: str) -> bool: ...

ErpCreateResult(order_number, created, details) carries your ERP's reference. Raise ErpRejected("text") for business refusals — the text goes to the operator; let anything else propagate — it becomes a generic failure and AIOTIC rolls the order back.

Template 1: functional API

Your ERP has an API that validates and creates orders (Business Central, Exact, Odoo, SAP B1 Service Layer, most SaaS ERPs). Copy aiotic/erp/functional_api.py, rename, and replace the three _map_* methods and the two HTTP calls:

class BusinessCentralAdapter(FunctionalApiAdapter):
    def _map_header(self, request_id, o):
        return {
            "customerNumber": o.customer.customer_id,
            "externalDocumentNumber": o.order_number[:35],
            "requestedDeliveryDate": o.delivery_date,
            "currencyCode": o.currency,
            "shipToName": o.shipping_details.recipient.company,
            "shipToAddressLine1": o.shipping_details.recipient.address.street,
            "shipToPostCode": o.shipping_details.recipient.address.postal_code,
            "shipToCity": o.shipping_details.recipient.address.city,
            "shipToCountry": o.shipping_details.recipient.address.country,
            "salesOrderLines": [{"lineType": "Item", "lineObjectNumber": i.article_number, "quantity": i.quantity, "unitOfMeasureCode": i.unit} for i in o.items],
        }

The ERP's own error text (e.g. "The Item does not exist. No.='PROD-999'") is raised as ErpRejected and reaches the operator unchanged. A light pipeline (sanitizers, RequiredFields, NoDuplicateOrder) is still recommended.

Template 2: data API / direct database

Your ERP exposes tables (or a generic data API that maps to tables). aiotic/erp/data_api.py shows the pattern with DB-API 2.0:

• all SQL in one dictionary (SQL[...]) so table/column names are changed in one place;
• header and lines in one transaction, keyed on external_ref = request_id;
• product_exists, get_customer, is_blocked implemented against the same tables, so the adapter doubles as CatalogPort and CustomerPort and the pipeline can enforce what the ERP will not.

import pyodbc
from aiotic.erp.data_api import DataApiAdapter

class MyErp(DataApiAdapter):
    SQL = {**DataApiAdapter.SQL,
           "item_exists": "SELECT 1 FROM dbo.Item WHERE No_ = ? AND Blocked = 0",
           "customer": "SELECT No_, Name, Blocked, [Credit Limit (LCY)] FROM dbo.Customer WHERE No_ = ?",
           ...}

erp = MyErp(lambda: pyodbc.connect(DSN), paramstyle="qmark", number_prefix="SO")

For the quick start, sqlite_demo() returns a connect factory to a seeded SQLite database with the three example tables.

With a data API the pipeline is mandatory

Nothing else validates. Run at least CustomerResolved, ArticlesInCatalog, PositiveQuantities, CustomerNotBlocked and NoDuplicateOrder before create_sales_order. See Validation when your ERP has no functional API.

Template 0: in-memory (tests, demos)

InMemoryErp(products=set(), customers={...}, blocked=set()) implements all three ports and behaves like a functional ERP (rejects unknown items). Used by the quick start and the SDK tests.

Idempotency inside the adapter

find_order_by_request_id is the second line of defence after the receiver's IdempotencyStore. Implement it against the external-reference column; it saves you when the store was wiped or when several service instances share an ERP but not a store.

Asynchronous ERPs

If order creation is slow (minutes), accept synchronously — validate, write a queued record with a provisional number, return success: true — and let a worker finish. success: true must mean "this will be booked"; if it later cannot be, that is an exception path for humans, not something AIOTIC will retry.