範例程式¶

以下為 Python（requests）範例，示範帶 API 金鑰推送交易（含重試）、送出日結與查詢對帳結果。

安裝¶

pip install requests

設定¶

API 金鑰請放在環境變數，不要寫在程式碼中。

export POSINGEST_API_KEY="<api_key>"
export POSINGEST_BASE_URL="https://func-posingest-dev.azurewebsites.net/pos/v1"

程式¶

import os
import time
import uuid

import requests

API_KEY = os.environ["POSINGEST_API_KEY"]
BASE_URL = os.environ["POSINGEST_BASE_URL"]

MAX_RETRY_SECONDS = 24 * 60 * 60  # 最多重試 24 小時


def send(method, path, payload=None):
    """送出請求；5xx／429／逾時以同一份 payload（同一 batch_id）重試。"""
    started = time.time()
    backoff_minutes = 1
    while True:
        try:
            resp = requests.request(
                method,
                f"{BASE_URL}{path}",
                json=payload,
                headers={"x-functions-key": API_KEY},
                timeout=60,
            )
        except requests.RequestException:
            resp = None

        if resp is not None:
            if resp.status_code in (200, 202):
                return resp.json()
            if resp.status_code == 429:
                time.sleep(int(resp.headers.get("Retry-After", "60")))
                continue
            if resp.status_code < 500:
                # 4xx：不重試，修正資料後以新的 batch_id 重送
                raise RuntimeError(f"{resp.status_code} {resp.text}")

        if time.time() - started > MAX_RETRY_SECONDS:
            raise RuntimeError("retry timeout")
        time.sleep(backoff_minutes * 60)
        backoff_minutes = min(backoff_minutes * 2, 60)


def push_trade_records(store_id, business_date, trade_records, batch_size=500):
    """分批推送交易，回傳批次數。"""
    batches = [trade_records[i:i + batch_size] for i in range(0, len(trade_records), batch_size)]
    for seq, records in enumerate(batches, start=1):
        payload = {
            "batch_id": str(uuid.uuid4()),  # 產生後在整個重試過程沿用
            "store_id": store_id,
            "business_date": business_date,
            "batch_seq": seq,
            "trade_records": records,
        }
        result = send("POST", "/trade-records", payload)
        for w in result.get("warnings", []):
            print("warning:", w)
    return len(batches)


def push_daily_close(store_id, business_date, control_totals):
    payload = {
        "batch_id": str(uuid.uuid4()),
        "store_id": store_id,
        "business_date": business_date,
        **control_totals,
    }
    return send("POST", "/daily-close", payload)


def wait_for_reconciliation(store_id, business_date, timeout_minutes=120):
    deadline = time.time() + timeout_minutes * 60
    while time.time() < deadline:
        result = send("GET", f"/daily-close/{store_id}/{business_date}")
        if result["status"] in ("MATCHED", "MISMATCHED"):
            return result
        time.sleep(10 * 60)
    raise RuntimeError("reconciliation not finished")


if __name__ == "__main__":
    store_id, business_date = "US001", "2026-10-14"
    trade_records = []  # 從 POS 讀出的交易，格式見「交易推送」

    batch_count = push_trade_records(store_id, business_date, trade_records)
    push_daily_close(store_id, business_date, {
        "batch_count": batch_count,
        "trade_record_count": len(trade_records),
        "item_row_count": sum(len(t["items"]) for t in trade_records),
        "payment_row_count": sum(len(t["payments"]) for t in trade_records),
        "net_sales_amount": "0.00",       # 依「送出日結」的計算定義
        "tax_amount": "0.00",
        "tip_amount": "0.00",
        "payment_total_amount": "0.00",
        "currency": "USD",
        "closed_at": "2026-10-15T00:12:00-07:00",
    })

    time.sleep(30 * 60)  # 送出日結約 30 分鐘後查詢
    result = wait_for_reconciliation(store_id, business_date)
    print(result["status"], result.get("differences"))

說明¶

send() 對 5xx、429、網路逾時以同一份 payload重試，因此 batch_id 不變，不會重複寫入。
4xx 錯誤直接拋出，需修正資料後以新的 batch_id 重送。
金額請以字串計算與傳送（Python 可用 decimal.Decimal），不要用 float。
正式環境建議將已送出的 batch_id 與結果記錄下來，程式中斷後可從中斷處繼續。