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

The contract has two halves. One receiver instance handles deliveries with the same request_id one at a time, remembers results in its IdempotencyStore and asks find_order_by_request_id before every create. Everything else is the ERP's: create_sales_order must be safe to call twice with the same request_id. Store the id as a unique external reference and, when the ERP reports that it already exists, return that order with created=False. FunctionalApiAdapter expects a 409 for a duplicate reference and looks the order up; DataApiAdapter relies on the unique column and resolves the conflict the same way. Two service instances, a wiped store, or an ERP that booked the order while the answer was lost are covered only by that uniqueness; no client-side code can replace it.

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.