تخطَّ إلى المحتوى
SlfloSlfloمستندات Slflo

Build a pull connector

هذا المحتوى غير متوفر بلغتك بعد.

A pull connector is a job your ERP runs every few minutes. Each run reads the orders that reps took since the last run, books one sales order for each, and tells Slflo what it did. Done this way, nothing is missed and nothing is booked twice — even when the job dies halfway, the network drops an answer, or the office sends a refused order again.

every 5 minutes:  pull new orders → stage them → book each once → ack each

Before you start: a token with orders:read and orders:write (sandbox), and the company in ERP mode API — only then do pre-sale orders wait for your ERP as ERP_PENDING.

The code on this page is from samples/python/pull_orders.py and samples/csharp/PullOrders.cs. Our build runs both against a real Slflo on every change, so what you copy is what we test. Both use the small client from Retries and token rotation.

  1. Keep a staging table with a unique key on (id, version, line_number).

    An order that changes while you read comes back later with a higher version; that is a new row. The same page read twice — because the job crashed before saving its watermark — inserts nothing. The unique key is what makes re-reading harmless, so put it in the database, not in your code.

    pull_orders.py — staging
    def open_staging():
        os.makedirs(STATE_DIR, exist_ok=True)
        db = sqlite3.connect(os.path.join(STATE_DIR, "staging.db"))
        db.executescript(
            """
            create table if not exists watermark (stream text primary key, version integer not null);
            create table if not exists order_line (
                order_id text not null,
                version integer not null,
                line_number integer not null,
                document text not null,          -- the order as pulled
                booked integer not null default 0,
                primary key (order_id, version, line_number)   -- the unique key that makes re-reads harmless
            );
            """
        )
        return db
    

    

    def stage(db, order):
        """One row per line under (id, version, line_number); a second copy inserts nothing."""
        lines = [line["line_number"] for line in order["lines"]] or [0]
        db.executemany(
            "insert or ignore into order_line (order_id, version, line_number, document) values (?, ?, ?, ?)",
            [(order["id"], order["version"], n, json.dumps(order)) for n in lines],
        )

  2. Pull from your watermark, only orders waiting for your ERP.

    Ask for filter[status]=ERP_PENDING after the highest version you have staged, and follow meta.next_version_after until meta.has_more is false. Save the watermark only once its page is staged.

    pull_orders.py — pull
    def pull(api, db, recover=False):
        """Stream style: from the watermark, only orders waiting for the ERP.
        Recover: from zero, only orders not acked at their current version."""
        query = {"filter[ack]": "pending"} if recover else {"filter[status]": "ERP_PENDING"}
        pulled = 0
        for page in api.pages("/orders", query, after=0 if recover else watermark(db)):
            for order in page["data"]:
                stage(db, order)
                pulled += 1
            if not recover:
                save_watermark(db, page["meta"]["next_version_after"])
            db.commit()  # the watermark moves only once its page is staged
        return pulled

  3. Book one sales order per Slflo order — find before you create.

    Put the Slflo order id in a field of the sales order you can search by (in Dynamics 365, CustomersOrderReference). Before creating, look for a sales order that already carries it. If the job died after creating the sales order but before the ack, the next run finds it instead of creating a twin. Take the account from customer.external_ref — your own key — and each line’s item from product.external_ref.

    pull_orders.py — book and ack
    def book(api, db, erp):
        result = {"created": 0, "found": 0, "accepted": 0, "rejected": 0, "already_answered": 0}
        staged = db.execute(
            "select distinct order_id, version, document from order_line where booked = 0 order by version"
        ).fetchall()
        for order_id, version, document in staged:
            order = json.loads(document)
            account = order["customer"]["external_ref"] or order["customer"]["code"]
            if fake_erp.on_hold(account):
                answer = {"outcome": "REJECTED", "reason_code": "CUSTOMER_ON_HOLD",
                          "message": f"Customer {account} is on hold in the ERP"}
                result["rejected"] += 1
            else:
                # Find before create: a run that died after creating the sales order but
                # before the ack finds it here instead of creating a twin.
                number = fake_erp.find_sales_order(erp, order["id"])
                if number:
                    result["found"] += 1
                else:
                    number = fake_erp.create_sales_order(erp, order["id"], account, order["totals"]["gross"])
                    result["created"] += 1
                answer = {"outcome": "ACCEPTED", "external_ref": number, "external_number": number}
                result["accepted"] += 1
            try:
                # The key names the order and the version answered, so a retry is a replay
                # and an order sent again (a new version) gets a fresh answer.
                api.post(f"/orders/{order_id}/ack", answer, key=f"{order_id}@{version}")
            except ApiError as refused:
                if refused.code not in ("ALREADY_ACKNOWLEDGED", "ORDER_NOT_AWAITING_ACK"):
                    raise
                result["already_answered"] += 1  # answered elsewhere, or no longer waiting
            db.execute("update order_line set booked = 1 where order_id = ? and version = ?", (order_id, version))
            db.commit()
        return result

  4. Ack every order you booked — or refused.

    ACCEPTED with your sales order number moves the order to ERP_CONFIRMED and stores your number on it. REJECTED moves it to ERP_FAILED and shows your message to the office. Use "<order id>@<version>" as the Idempotency-Key: a retry of a lost answer replays, and an order sent again (a new version) gets a fresh answer.

    POST /orders/ord_01J9X/ack

    نافذة طرفيّة
    curl -X POST 'https://api.slflo.com/api/integration/v1/orders/ord_01J9X/ack' \
      -H 'Authorization: Bearer $SLFLO_TOKEN' \
      -H 'Idempotency-Key: ord_01J9X@18342' \
      -H 'Content-Type: application/json' \
      -d '{"outcome":"ACCEPTED","external_ref":"SO-0012345","external_number":"SO-0012345"}'

    Response

    HTTP 200
    

    {
      "data": {
        "id": "ord_01J9X",
        "number": "SO-000123",
        "status": "ERP_CONFIRMED",
        "version": 18343,
        "result": "applied",
        "ack": {
          "outcome": "ACCEPTED",
          "external_ref": "SO-0012345",
          "external_number": "SO-0012345",
          "acked_version": 18343,
          "acked_at": "2026-10-05T10:42:00Z"
        }
      }
    }

Run it

Section titled “Run it”
Terminal window
export SLFLO_TOKEN='sfi_live_…' SLFLO_BASE='https://api.slflo.com/api/integration/v1'
python3 pull_orders.py

The last line is what the run did. With two orders waiting for your ERP:

{"accepted": 2, "already_answered": 0, "created": 2, "found": 0, "mode": "run", "pulled": 2, "rejected": 0}

Run it again straight away and it pulls nothing and books nothing. On the hosted sandbox the first run pulls nothing either — the day’s two orders are already delivered — until we place a new order for you (what needs us).

A refused order, sent again, then accepted

Section titled “A refused order, sent again, then accepted”

Your ERP will sometimes refuse an order — the account is on credit hold, an item is blocked. Refuse it with a reason the office can act on; don’t leave it unanswered.

  1. Your run acks it REJECTED with a reason_code and a message:

    POST /orders/ord_01J9X/ack

    نافذة طرفيّة
    curl -X POST 'https://api.slflo.com/api/integration/v1/orders/ord_01J9X/ack' \
      -H 'Authorization: Bearer $SLFLO_TOKEN' \
      -H 'Idempotency-Key: ord_01J9X@18342' \
      -H 'Content-Type: application/json' \
      -d '{"outcome":"REJECTED","reason_code":"CUSTOMER_ON_HOLD","message":"Customer CUST-1002 is on hold in the ERP"}'

    Response

    HTTP 200
    

    {
      "data": {
        "id": "ord_01J9X",
        "number": "SO-000123",
        "status": "ERP_FAILED",
        "version": 18343,
        "result": "applied",
        "ack": {
          "outcome": "REJECTED",
          "reason_code": "CUSTOMER_ON_HOLD",
          "message": "Customer CUST-1002 is on hold in the ERP",
          "external_ref": null,
          "external_number": null,
          "acked_version": 18343,
          "acked_at": "2026-10-05T10:42:00Z"
        }
      }
    }

  2. On the order page the office sees Refused by the ERP with your message. Once the cause is fixed (the hold lifted), someone allowed to deliver orders chooses Send to ERP again.

  3. The order is ERP_PENDING again, at a higher version, so your next pull sees it.

  4. Your run books it — find-before-create finds nothing, because the refusal created nothing — and acks it ACCEPTED. The key id@version is new, so this is a new answer, not a replay.

Try it with the stand-in ERP: put the order’s account on hold, then lift it. ERP_ON_HOLD takes accounts, comma-separated — the customer’s external_ref, or its Slflo code when your ERP hasn’t pushed it. The middle step happens in the console — on the hosted sandbox, ask us to send the order again.

Terminal window
ERP_ON_HOLD=CUST-1002 python3 pull_orders.py   # {"rejected": 1, ...}
# the office: Send to ERP again
python3 pull_orders.py                         # {"pulled": 1, "created": 1, ...}

When it goes wrong

Section titled “When it goes wrong”
What you see Why What to do
409 ERP_MODE_NOT_API on every ack The company isn’t in ERP mode API, so orders don’t wait for you Ask us to switch it; until then read invoices instead (by seq)
No orders arrive, though reps sell Van sales never wait for the ERP — they’re delivered and invoiced on the spot Read them as invoices; only pre-sale orders come through here
409 ALREADY_ACKNOWLEDGED You answered this order before, differently Treat it as done. A new answer is taken only after the office sends it again
409 ORDER_NOT_AWAITING_ACK Someone answered it (another job, a manual call), or it was cancelled Nothing to book. Pull filter[ack]=pending for what still waits
409 EXTERNAL_REF_CONFLICT Your sales order number is already on another Slflo order Each record answers to one number of yours. Look for two jobs writing to the same sequence
401 mid-run The token was revoked or expired Stop and alert. The next run with a good token finds what this one created (rotation)
The job died after creating sales orders Power cut, deployment Just run again: find-before-create finds them and acks them
Staging purged, or the ERP database restored The watermark points past orders you never booked Recover from filter[ack]=pending

The rest of the sample

Section titled “The rest of the sample”

The stand-in for your ERP, which you replace with your ERP’s own calls:

fake_erp.py
"""A stand-in for your ERP's books, so the samples run end to end.


Replace these functions with your ERP's own calls (D365 services, SAP BAPIs, Odoo RPC ...).
It keeps sales orders and journals in STATE_DIR/erp.db. Accounts listed in ERP_ON_HOLD
(comma-separated) are refused, the way a credit department's hold refuses them.
"""
import os
import sqlite3


STATE_DIR = os.environ.get("STATE_DIR", "state")




def connect():
    os.makedirs(STATE_DIR, exist_ok=True)
    db = sqlite3.connect(os.path.join(STATE_DIR, "erp.db"))
    db.executescript(
        """
        create table if not exists sales_order (
            number integer primary key autoincrement,
            customer_reference text not null unique,   -- the Slflo order id
            account text not null,
            total text not null
        );
        create table if not exists journal (
            number integer primary key autoincrement,
            kind text not null,                       -- invoice, payment, return, payment-reversal
            source_id text not null,
            source_version integer not null,
            seq integer,
            amount text not null,
            unique (kind, source_id, source_version)
        );
        """
    )
    return db




def on_hold(account):
    return account in {a.strip() for a in os.environ.get("ERP_ON_HOLD", "").split(",") if a.strip()}




def find_sales_order(db, customer_reference):
    row = db.execute("select number from sales_order where customer_reference = ?", (customer_reference,)).fetchone()
    return f"SO-{row[0]:07d}" if row else None




def create_sales_order(db, customer_reference, account, total):
    cursor = db.execute(
        "insert into sales_order (customer_reference, account, total) values (?, ?, ?)",
        (customer_reference, account, total),
    )
    db.commit()
    return f"SO-{cursor.lastrowid:07d}"




def post_journal(db, kind, source_id, source_version, amount, seq=None):
    """One journal per document version; posting the same version again returns the first."""
    row = db.execute(
        "select number from journal where kind = ? and source_id = ? and source_version = ?",
        (kind, source_id, source_version),
    ).fetchone()
    if row is None:
        row = (
            db.execute(
                "insert into journal (kind, source_id, source_version, seq, amount) values (?, ?, ?, ?, ?)",
                (kind, source_id, source_version, seq, amount),
            ).lastrowid,
        )
        db.commit()
    return f"JRN-{row[0]:06d}"




def journal_of(db, kind, source_id):
    row = db.execute(
        "select number from journal where kind = ? and source_id = ? order by source_version limit 1",
        (kind, source_id),
    ).fetchone()
    return f"JRN-{row[0]:06d}" if row else None




def booked_seqs(db, kind):
    return [r[0] for r in db.execute("select distinct seq from journal where kind = ? and seq is not null order by seq", (kind,))]

The complete connector — orders, invoices, payments, returns and master data in one program — is the Kotlin reference connector.