The SAP data model for AI engineers: tables, CDS views and OData

"""The SAP data model for AI engineers: trace one blocked sales order and write a data dictionary.

How to run (from the unit01 folder, with the course .venv turned on):
  python data_model_trace.py --sample           made-up data on your computer; no account, no internet
  python data_model_trace.py                    SAP Business Accelerator Hub sandbox (needs SAP_API_KEY in .env)
  python data_model_trace.py --order 1234567    trace this order instead of the first blocked one

The script only reads. It writes two files next to itself:
  blocked_order_trace.json   the order, its items, its partners and the sold-to customer
  data_dictionary.md         what every field in that file means, taken from the APIs' metadata
"""
import argparse
import json
import os
import re
import sys
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path

SANDBOX = "https://sandbox.api.sap.com/s4hanacloud"
SERVICES = {
    "sales": "/sap/opu/odata/sap/API_SALES_ORDER_SRV",
    "bp": "/sap/opu/odata/sap/API_BUSINESS_PARTNER",
}

# What we read from each entity set: (service, entity set, fields, one-line purpose).
# Every field name must exist in that service's $metadata; step A checks it.
PLAN = [
    ("sales", "A_SalesOrder",
     ["SalesOrder", "SalesOrderType", "SalesOrganization", "SoldToParty", "CreationDate",
      "RequestedDeliveryDate", "TotalNetAmount", "TransactionCurrency", "DeliveryBlockReason",
      "HeaderBillingBlockReason", "TotalCreditCheckStatus"],
     "Sales order header: one record per order. Who ordered, when, how much, and any blocks."),
    ("sales", "A_SalesOrderItem",
     ["SalesOrder", "SalesOrderItem", "Material", "SalesOrderItemText", "RequestedQuantity",
      "RequestedQuantityUnit", "NetAmount", "TransactionCurrency", "ItemBillingBlockReason"],
     "Sales order item: one record per line of an order. What was ordered and how much."),
    ("sales", "A_SalesOrderHeaderPartner",
     ["SalesOrder", "PartnerFunction", "Customer"],
     "Partners of the order: one record per role (sold-to, ship-to and others)."),
    ("bp", "A_Customer",
     ["Customer", "CustomerName", "OrderIsBlockedForCustomer", "DeliveryIsBlocked",
      "BillingIsBlockedForCustomer"],
     "Customer master record: the customer's name and central blocks for all sales areas."),
    ("bp", "A_BusinessPartner",
     ["BusinessPartner", "Customer", "BusinessPartnerFullName", "BusinessPartnerCategory",
      "BusinessPartnerIsBlocked"],
     "Business partner: the central record that the customer role belongs to."),
]

# Rules the metadata cannot tell you. These go into the dictionary as notes.
CODE_NOTE = "A code, not text. Its meaning is set in each company's configuration. Empty means none."
NOTES = {
    "DeliveryBlockReason": CODE_NOTE,
    "HeaderBillingBlockReason": CODE_NOTE,
    "ItemBillingBlockReason": CODE_NOTE,
    "TotalCreditCheckStatus": "A status code. Look up its values for your release; don't guess.",
    "OrderIsBlockedForCustomer": CODE_NOTE,
    "DeliveryIsBlocked": CODE_NOTE,
    "BillingIsBlockedForCustomer": CODE_NOTE,
    "PartnerFunction": "A code for the partner's role. Check the codes your system returns.",
    "SalesOrderType": "A code defined in configuration.",
    "SalesOrganization": "An organizational unit key defined in configuration.",
}


# ------------------------------------------------------------------ talking to the API

class LiveSAP:
    """Sends real HTTP GET requests to SAP's sandbox."""

    def __init__(self, key: str):
        import requests  # imported here so --sample works even without the library
        self.requests = requests
        self.session = requests.Session()
        self.session.headers.update({"APIKey": key})

    def get(self, service: str, path: str, params: dict | None = None) -> tuple[int, str]:
        accept = "application/xml" if path == "/$metadata" else "application/json"
        try:
            resp = self.session.get(SANDBOX + SERVICES[service] + path, params=params,
                                    headers={"Accept": accept}, timeout=30)
        except self.requests.exceptions.RequestException as err:
            sys.exit(f"Could not reach sandbox.api.sap.com ({type(err).__name__}). "
                     "Check your network or proxy, or run with --sample.")
        return resp.status_code, resp.text


class SampleSAP:
    """Answers like the sandbox would, from made-up data. Understands $top, $select and a
    simple $filter of the form  Field eq 'value', which is all this script uses."""

    def get(self, service: str, path: str, params: dict | None = None) -> tuple[int, str]:
        params = params or {}
        if path == "/$metadata":
            return 200, SAMPLE_METADATA[service]
        rows = None
        one = re.fullmatch(r"/(\w+)\('([^']*)'\)", path)
        nav = re.fullmatch(r"/A_SalesOrder\('(\d+)'\)/(to_Item|to_Partner)", path)
        if nav:
            source = SAMPLE_ITEMS if nav.group(2) == "to_Item" else SAMPLE_PARTNERS
            rows = source.get(nav.group(1), [])
        elif one and one.group(1) in SAMPLE_SETS[service]:
            hits = [r for r in SAMPLE_SETS[service][one.group(1)]
                    if r[SAMPLE_KEYS[one.group(1)]] == one.group(2)]
            if not hits:
                return 404, json.dumps({"error": {"message": {"value": "Resource not found"}}})
            return 200, json.dumps({"d": select(hits[0], params)})
        elif path.lstrip("/") in SAMPLE_SETS[service]:
            rows = SAMPLE_SETS[service][path.lstrip("/")]
        if rows is None:
            return 404, json.dumps({"error": {"message": {"value": f"Resource not found: {path}"}}})
        if "$filter" in params:
            field, value = re.fullmatch(r"(\w+) eq '([^']*)'", params["$filter"]).groups()
            rows = [r for r in rows if r.get(field) == value]
        rows = rows[:int(params.get("$top", len(rows)))]
        return 200, json.dumps({"d": {"results": [select(r, params) for r in rows]}})


def select(row: dict, params: dict) -> dict:
    if "$select" not in params:
        return row
    return {k: row[k] for k in params["$select"].split(",") if k in row}


# ------------------------------------------------------------------ step A: the contract

def local(name: str) -> str:
    """'API_SALES_ORDER_SRV.A_SalesOrderType' -> 'A_SalesOrderType'; '{ns}Property' -> 'Property'."""
    return name.rsplit("}", 1)[-1].rsplit(".", 1)[-1]


def sap_attr(node, name: str) -> str:
    """Read an SAP annotation such as sap:label, whatever namespace prefix the file uses."""
    return next((v for k, v in node.attrib.items() if k.endswith("}" + name)), "")


def read_contract(xml_text: str) -> dict:
    """Parse OData V2 $metadata into: entity set -> its key, properties and navigation."""
    root = ET.fromstring(xml_text.encode("utf-8"))
    types, ends, sets = {}, {}, {}
    for node in root.iter():
        tag = local(node.tag)
        if tag == "EntityType":
            key = [ref.get("Name") for ref in node.iter() if local(ref.tag) == "PropertyRef"]
            props, navs = {}, {}
            for child in node:
                if local(child.tag) == "Property":
                    props[child.get("Name")] = {
                        "type": child.get("Type", "").replace("Edm.", ""),
                        "max": child.get("MaxLength", ""),
                        "label": sap_attr(child, "label"),
                        "quickinfo": sap_attr(child, "quickinfo"),
                        "unit": sap_attr(child, "unit"),
                    }
                elif local(child.tag) == "NavigationProperty":
                    navs[child.get("Name")] = (local(child.get("Relationship", "")),
                                               child.get("ToRole"))
            types[node.get("Name")] = {"key": key, "props": props, "navs": navs}
        elif tag == "Association":
            for end in node:
                if local(end.tag) == "End":
                    ends[(node.get("Name"), end.get("Role"))] = (local(end.get("Type")),
                                                                 end.get("Multiplicity"))
        elif tag == "EntitySet":
            sets[node.get("Name")] = local(node.get("EntityType"))
    set_of_type = {t: s for s, t in sets.items()}
    contract = {}
    for set_name, type_name in sets.items():
        t = types[type_name]
        navigation = {}
        for nav_name, (assoc, role) in t["navs"].items():
            target_type, many = ends.get((assoc, role), ("?", "?"))
            navigation[nav_name] = {"target": set_of_type.get(target_type, target_type),
                                    "many": many == "*"}
        contract[set_name] = {"type": type_name, "key": t["key"], "props": t["props"],
                              "navigation": navigation}
    return contract


# ------------------------------------------------------------------ steps B to E: the data

def get_json(api, service: str, path: str, params: dict | None = None):
    status, text = api.get(service, path, params)
    if status != 200:
        hints = {401: "the key is missing or wrong: check SAP_API_KEY in .env",
                 403: "the key is valid but not allowed to do this",
                 404: "the address is wrong, or the record does not exist",
                 429: "too many calls: wait a minute and run again"}
        sys.exit(f"Stopped: HTTP {status}, {hints.get(status, 'see the message below')}\n{text[:300]}")
    data = json.loads(text)["d"]
    return data.get("results", data) if isinstance(data, dict) else data


def fields_of(entity_set: str) -> str:
    return ",".join(next(p[2] for p in PLAN if p[1] == entity_set))


def odata_date(value):
    """OData V2 JSON sends dates as /Date(milliseconds since 1970)/. Turn that into 2026-10-18."""
    match = re.search(r"/Date\((-?\d+)", value or "") if isinstance(value, str) else None
    if not match:
        return value
    return datetime.fromtimestamp(int(match.group(1)) / 1000, tz=timezone.utc).date().isoformat()


def tidy(row: dict) -> dict:
    """Keep API field names (the dictionary explains them); fix dates; drop __metadata."""
    return {k: odata_date(v) for k, v in row.items() if not k.startswith("__")}


def safe_key(value) -> str:
    """Keys go into URLs and $filter text, so allow only short plain codes (no quotes, slashes or spaces)."""
    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,10}", value):
        sys.exit(f"Stopped: {value!r} is not a valid key (letters, digits, - or _, at most 10).")
    return value


def is_blocked(order: dict) -> bool:
    return bool(order.get("DeliveryBlockReason") or order.get("HeaderBillingBlockReason"))


def find_order(api, wanted: str | None, scan: int) -> dict:
    """Step B: the header. Either the order you asked for, or the first blocked one we find."""
    if wanted:
        return get_json(api, "sales", f"/A_SalesOrder('{safe_key(wanted)}')", {"$select": fields_of("A_SalesOrder")})
    orders = get_json(api, "sales", "/A_SalesOrder",
                      {"$top": str(scan), "$select": fields_of("A_SalesOrder"), "$orderby": "SalesOrder"})
    blocked = [o for o in orders if is_blocked(o)]
    print(f"   Scanned {len(orders)} orders; with a delivery or billing block: {len(blocked)}")
    if blocked:
        return blocked[0]
    if not orders:
        sys.exit("   The service returned no orders. Nothing to trace.")
    print("   No blocked order in this slice. That is a valid answer; tracing the first order instead.")
    return orders[0]


# ------------------------------------------------------------------ step F: the dictionary

def describe(prop: dict) -> str:
    """Prefer the longer tooltip text (sap:quickinfo); fall back to the short label."""
    return prop["quickinfo"] or prop["label"] or "(no label in metadata)"


def build_dictionary(contracts: dict, source: str) -> str:
    today = datetime.now(timezone.utc).date().isoformat()
    lines = [
        "# Data dictionary: blocked sales order trace",
        "",
        f"Generated {today} from the $metadata of API_SALES_ORDER_SRV and API_BUSINESS_PARTNER "
        f"({source}). Field names are the exact API names used in blocked_order_trace.json.",
        "",
        "## How the records connect",
        "",
        "- A_SalesOrder --to_Item--> A_SalesOrderItem (many). Item key: SalesOrder + SalesOrderItem.",
        "- A_SalesOrder --to_Partner--> A_SalesOrderHeaderPartner (many). One record per partner role.",
        "- A_SalesOrderHeaderPartner.Customer = A_Customer.Customer (a different API: API_BUSINESS_PARTNER).",
        "- A_Customer.Customer = A_BusinessPartner.Customer. The business partner number may differ.",
        "",
        "## Rules",
        "",
        "- Amounts are decimals sent as text. Read the currency from the field named in Notes.",
        "- Codes are not text. Do not guess what a code means; say the code and that it needs looking up.",
        "- Empty text means \"not set\", not a code.",
    ]
    for service, entity_set, fields, purpose in PLAN:
        c = contracts[service][entity_set]
        lines += ["", f"## {entity_set}", "", purpose, f"Key: {' + '.join(c['key'])}.", "",
                  "| Field | Meaning (from SAP metadata) | Type | Notes |", "| --- | --- | --- | --- |"]
        for name in fields:
            p = c["props"][name]
            typ = p["type"] + (f"({p['max']})" if p["max"] else "")
            notes = []
            if name in c["key"]:
                notes.append("Key.")
            if p["unit"]:
                notes.append(f"Unit or currency in {p['unit']}.")
            if name in NOTES:
                notes.append(NOTES[name])
            lines.append(f"| {name} | {describe(p)} | {typ} | {' '.join(notes)} |")
    return "\n".join(lines) + "\n"


# ------------------------------------------------------------------ main

def main() -> None:
    parser = argparse.ArgumentParser(description="Trace one blocked sales order and write a data dictionary.")
    parser.add_argument("--sample", action="store_true", help="use made-up data; no key or internet")
    parser.add_argument("--order", help="trace this sales order number instead of the first blocked one")
    parser.add_argument("--scan", type=int, default=100, help="how many orders to scan for a blocked one")
    args = parser.parse_args()

    if args.sample:
        api, source = SampleSAP(), "sample data"
        print("Target: sample data on your computer (made up, shaped like SAP's APIs)\n")
    else:
        try:
            from dotenv import load_dotenv
            load_dotenv()
        except ImportError:
            pass
        key = os.environ.get("SAP_API_KEY")
        if not key:
            sys.exit("SAP_API_KEY is not set. Add it to .env, or run with --sample.")
        api, source = LiveSAP(key), "SAP sandbox"
        print("Target: SAP Business Accelerator Hub sandbox\n")

    print("A. Read both contracts ($metadata)")
    contracts = {}
    for service in SERVICES:
        status, text = api.get(service, "/$metadata")
        if status != 200:
            sys.exit(f"   Could not read $metadata of {SERVICES[service]}: HTTP {status}")
        contracts[service] = read_contract(text)
    for service, entity_set, fields, _ in PLAN:
        c = contracts[service].get(entity_set)
        if not c:
            sys.exit(f"   {entity_set} is not in {SERVICES[service]}. Check the entity set name.")
        missing = [f for f in fields if f not in c["props"]]
        if missing:
            sys.exit(f"   {entity_set} has no field {', '.join(missing)}. Check the field list.")
        print(f"   {entity_set:<26} key {'+'.join(c['key']):<26} {len(c['props']):>3} fields, "
              f"{len(c['navigation'])} navigation properties")
    for nav in ("to_Item", "to_Partner"):
        n = contracts["sales"]["A_SalesOrder"]["navigation"][nav]
        print(f"   A_SalesOrder --{nav}--> {n['target']} ({'many' if n['many'] else 'one'})")

    print("\nB. Find the order (header)")
    order = tidy(find_order(api, args.order, args.scan))
    oid = safe_key(order["SalesOrder"])
    print(f"   Order {oid}: sold-to {order.get('SoldToParty')}, "
          f"{order.get('TotalNetAmount')} {order.get('TransactionCurrency')}, "
          f"delivery block '{order.get('DeliveryBlockReason')}', "
          f"billing block '{order.get('HeaderBillingBlockReason')}'")

    print("\nC. Follow to_Item (what was ordered)")
    items = [tidy(i) for i in get_json(api, "sales", f"/A_SalesOrder('{oid}')/to_Item",
                                       {"$select": fields_of("A_SalesOrderItem")})]
    for i in items:
        print(f"   item {i['SalesOrderItem']}: {i.get('RequestedQuantity')} {i.get('RequestedQuantityUnit')} "
              f"{i.get('Material')} ({i.get('SalesOrderItemText')})")

    print("\nD. Follow to_Partner (who is involved)")
    partners = [tidy(p) for p in get_json(api, "sales", f"/A_SalesOrder('{oid}')/to_Partner",
                                          {"$select": fields_of("A_SalesOrderHeaderPartner")})]
    for p in partners:
        print(f"   partner function {p['PartnerFunction']}: customer {p.get('Customer') or '-'}")

    print("\nE. Cross into the Business Partner API (the sold-to customer)")
    sold_to = safe_key(order.get("SoldToParty"))
    customer = tidy(get_json(api, "bp", f"/A_Customer('{sold_to}')", {"$select": fields_of("A_Customer")}))
    print(f"   A_Customer {sold_to}: {customer.get('CustomerName')}; central blocks: "
          f"order '{customer.get('OrderIsBlockedForCustomer')}', "
          f"delivery '{customer.get('DeliveryIsBlocked')}', billing '{customer.get('BillingIsBlockedForCustomer')}'")
    bps = get_json(api, "bp", "/A_BusinessPartner",
                   {"$filter": f"Customer eq '{sold_to}'", "$select": fields_of("A_BusinessPartner")})
    bp = tidy(bps[0]) if bps else None
    if bp:
        print(f"   A_BusinessPartner {bp['BusinessPartner']}: {bp.get('BusinessPartnerFullName')}, "
              f"category {bp.get('BusinessPartnerCategory')}, centrally blocked: {bp.get('BusinessPartnerIsBlocked')}")
    else:
        print("   No business partner points to this customer number. Valid, but worth asking about.")

    print("\nF. Save the trace and the data dictionary")
    here = Path(__file__).parent
    trace = {"source": source, "sales_order": order, "items": items, "partners": partners,
             "sold_to": {"customer": customer, "business_partner": bp}}
    (here / "blocked_order_trace.json").write_text(json.dumps(trace, indent=2), encoding="utf-8")
    dictionary = build_dictionary(contracts, source)
    (here / "data_dictionary.md").write_text(dictionary, encoding="utf-8")
    print(f"   Wrote blocked_order_trace.json ({len(items)} items, {len(partners)} partners)")
    print(f"   Wrote data_dictionary.md ({len(dictionary.splitlines())} lines, {len(dictionary)} characters)")
    print("   First rows of the A_SalesOrder table in the dictionary:")
    start = dictionary.index("| Field |", dictionary.index("## A_SalesOrder"))
    for line in dictionary[start:].splitlines()[2:6]:
        print("   " + line)


# ------------------------------------------------------------------ made-up sample data
# Shaped like SAP's metadata: entity types end in "Type", entity sets point to them, and
# navigation properties point to associations. Labels follow SAP's published models.

def _prop(name, typ, label, quickinfo="", max_len="", unit=""):
    extra = f' MaxLength="{max_len}"' if max_len else ""
    extra += f' sap:quickinfo="{quickinfo}"' if quickinfo else ""
    extra += f' sap:unit="{unit}"' if unit else ""
    return f'<Property Name="{name}" Type="Edm.{typ}"{extra} sap:label="{label}"/>'


def _metadata(namespace, types, assocs):
    body = []
    for type_name, key, props, navs in types:
        body.append(f'<EntityType Name="{type_name}"><Key>'
                    + "".join(f'<PropertyRef Name="{k}"/>' for k in key) + "</Key>"
                    + "".join(props)
                    + "".join(f'<NavigationProperty Name="{n}" Relationship="{namespace}.{a}" '
                              f'FromRole="From_{a}" ToRole="To_{a}"/>' for n, a in navs)
                    + "</EntityType>")
    for name, src, dst, many in assocs:
        body.append(f'<Association Name="{name}"><End Type="{namespace}.{src}" Multiplicity="1" '
                    f'Role="From_{name}"/><End Type="{namespace}.{dst}" Multiplicity="{many}" '
                    f'Role="To_{name}"/></Association>')
    body.append('<EntityContainer Name="' + namespace + '_Entities">'
                + "".join(f'<EntitySet Name="{t[0][:-4]}" EntityType="{namespace}.{t[0]}"/>' for t in types)
                + "</EntityContainer>")
    return ('<?xml version="1.0" encoding="utf-8"?>'
            '<edmx:Edmx Version="1.0" xmlns:edmx="http://schemas.microsoft.com/ado/2007/06/edmx" '
            'xmlns:m="http://schemas.microsoft.com/ado/2007/08/dataservices/metadata" '
            'xmlns:sap="http://www.sap.com/Protocols/SAPData"><edmx:DataServices m:DataServiceVersion="2.0">'
            f'<Schema Namespace="{namespace}" xmlns="http://schemas.microsoft.com/ado/2008/09/edm">'
            + "".join(body) + "</Schema></edmx:DataServices></edmx:Edmx>")


SAMPLE_METADATA = {
    "sales": _metadata("API_SALES_ORDER_SRV", [
        ("A_SalesOrderType", ["SalesOrder"], [
            _prop("SalesOrder", "String", "Sales Order", max_len="10"),
            _prop("SalesOrderType", "String", "Sales Order Type", max_len="4"),
            _prop("SalesOrganization", "String", "Sales Organization", max_len="4"),
            _prop("SoldToParty", "String", "Sold-to Party", max_len="10"),
            _prop("CreationDate", "DateTime", "Created On", "Record Created On"),
            _prop("RequestedDeliveryDate", "DateTime", "Req. Deliv. Date", "Requested Delivery Date"),
            _prop("TotalNetAmount", "Decimal", "Net Value",
                  "Net Value of the Sales Order in Document Currency", unit="TransactionCurrency"),
            _prop("TransactionCurrency", "String", "Doc. Currency", "SD Document Currency", "5"),
            _prop("DeliveryBlockReason", "String", "Delivery Block", "Delivery Block (Document Header)", "2"),
            _prop("HeaderBillingBlockReason", "String", "Billing Block", "Billing Block in SD Document", "2"),
            _prop("TotalCreditCheckStatus", "String", "Credit Status", "Overall Status of Credit Checks", "1"),
        ], [("to_Item", "assoc_Item"), ("to_Partner", "assoc_Partner")]),
        ("A_SalesOrderItemType", ["SalesOrder", "SalesOrderItem"], [
            _prop("SalesOrder", "String", "Sales Order", max_len="10"),
            _prop("SalesOrderItem", "String", "Item", "Sales Order Item", "6"),
            _prop("Material", "String", "Material", "Material Number", "40"),
            _prop("SalesOrderItemText", "String", "Item Description", "Short Text for Sales Order Item", "40"),
            _prop("RequestedQuantity", "Decimal", "Requested Qty", "Requested Quantity",
                  unit="RequestedQuantityUnit"),
            _prop("RequestedQuantityUnit", "String", "Unit", "Unit of the Requested Quantity", "3"),
            _prop("NetAmount", "Decimal", "Net Value",
                  "Net Value of the Order Item in Document Currency", unit="TransactionCurrency"),
            _prop("TransactionCurrency", "String", "Doc. Currency", "SD Document Currency", "5"),
            _prop("ItemBillingBlockReason", "String", "Billing Block", "Billing Block for Item", "2"),
            _prop("DeliveryStatus", "String", "Delivery Status", "Delivery Status (Item)", "1"),
        ], [("to_SalesOrder", "assoc_ItemHeader")]),
        ("A_SalesOrderHeaderPartnerType", ["SalesOrder", "PartnerFunction"], [
            _prop("SalesOrder", "String", "Sales Order", max_len="10"),
            _prop("PartnerFunction", "String", "Partner Function", max_len="2"),
            _prop("Customer", "String", "Customer", "Customer Number", "10"),
        ], []),
    ], [("assoc_Item", "A_SalesOrderType", "A_SalesOrderItemType", "*"),
        ("assoc_Partner", "A_SalesOrderType", "A_SalesOrderHeaderPartnerType", "*"),
        ("assoc_ItemHeader", "A_SalesOrderItemType", "A_SalesOrderType", "1")]),
    "bp": _metadata("API_BUSINESS_PARTNER", [
        ("A_CustomerType", ["Customer"], [
            _prop("Customer", "String", "Customer", "Customer Number", "10"),
            _prop("CustomerName", "String", "Name of Customer", max_len="80"),
            _prop("OrderIsBlockedForCustomer", "String", "Order Block", "Central Order Block for Customer", "2"),
            _prop("DeliveryIsBlocked", "String", "Delivery block", "Central delivery block for the customer", "2"),
            _prop("BillingIsBlockedForCustomer", "String", "Billing Block", "Central Billing Block for Customer", "2"),
        ], []),
        ("A_BusinessPartnerType", ["BusinessPartner"], [
            _prop("BusinessPartner", "String", "Business Partner", "Business Partner Number", "10"),
            _prop("Customer", "String", "Customer", "Customer Number", "10"),
            '<Property Name="BusinessPartnerFullName" Type="Edm.String" MaxLength="81"/>',
            _prop("BusinessPartnerCategory", "String", "BP Category", "Business Partner Category", "1"),
            _prop("BusinessPartnerIsBlocked", "Boolean", "Central Block", "Central Block for Business Partner"),
        ], [("to_Customer", "assoc_BPCustomer")]),
    ], [("assoc_BPCustomer", "A_BusinessPartnerType", "A_CustomerType", "0..1")]),
}


def _order(no, party, created, amount, dblock="", bblock="", credit=""):
    return {"SalesOrder": no, "SalesOrderType": "OR", "SalesOrganization": "1710", "SoldToParty": party,
            "CreationDate": f"/Date({created})/", "RequestedDeliveryDate": f"/Date({created + 1209600000})/",
            "TotalNetAmount": amount, "TransactionCurrency": "USD", "DeliveryBlockReason": dblock,
            "HeaderBillingBlockReason": bblock, "TotalCreditCheckStatus": credit}


def _item(order, no, material, text, qty, amount, bblock=""):
    return {"SalesOrder": order, "SalesOrderItem": no, "Material": material, "SalesOrderItemText": text,
            "RequestedQuantity": qty, "RequestedQuantityUnit": "PC", "NetAmount": amount,
            "TransactionCurrency": "USD", "ItemBillingBlockReason": bblock, "DeliveryStatus": "A"}


SAMPLE_SETS = {
    "sales": {"A_SalesOrder": [
        _order("9000003", "CUST-C", 1789171200000, "5100.00"),
        _order("9000004", "CUST-A", 1789257600000, "7300.00", dblock="01", credit="B"),
        _order("9000005", "CUST-D", 1789344000000, "12000.00"),
    ]},
    "bp": {
        "A_Customer": [
            {"Customer": "CUST-A", "CustomerName": "Northwind Pumps Inc.", "OrderIsBlockedForCustomer": "",
             "DeliveryIsBlocked": "", "BillingIsBlockedForCustomer": ""},
            {"Customer": "CUST-C", "CustomerName": "Contoso Valves", "OrderIsBlockedForCustomer": "",
             "DeliveryIsBlocked": "", "BillingIsBlockedForCustomer": ""},
        ],
        "A_BusinessPartner": [
            {"BusinessPartner": "1000042", "Customer": "CUST-A", "BusinessPartnerFullName": "Northwind Pumps Inc.",
             "BusinessPartnerCategory": "2", "BusinessPartnerIsBlocked": False},
        ],
    },
}
SAMPLE_KEYS = {"A_SalesOrder": "SalesOrder", "A_Customer": "Customer", "A_BusinessPartner": "BusinessPartner"}
SAMPLE_ITEMS = {
    "9000004": [_item("9000004", "10", "VALVE-40", "Control valve", "4", "6800.00"),
                _item("9000004", "20", "SEAL-7", "Seal kit", "10", "500.00")],
}
SAMPLE_PARTNERS = {
    "9000004": [{"SalesOrder": "9000004", "PartnerFunction": "SP", "Customer": "CUST-A"},
                {"SalesOrder": "9000004", "PartnerFunction": "SH", "Customer": "CUST-A"}],
}

if __name__ == "__main__":
    main()

Target: sample data on your computer (made up, shaped like SAP's APIs)

A. Read both contracts ($metadata)
   A_SalesOrder               key SalesOrder                  11 fields, 2 navigation properties
   A_SalesOrderItem           key SalesOrder+SalesOrderItem   10 fields, 1 navigation properties
   A_SalesOrderHeaderPartner  key SalesOrder+PartnerFunction   3 fields, 0 navigation properties
   A_Customer                 key Customer                     5 fields, 0 navigation properties
   A_BusinessPartner          key BusinessPartner              5 fields, 1 navigation properties
   A_SalesOrder --to_Item--> A_SalesOrderItem (many)
   A_SalesOrder --to_Partner--> A_SalesOrderHeaderPartner (many)

B. Find the order (header)
   Scanned 3 orders; with a delivery or billing block: 1
   Order 9000004: sold-to CUST-A, 7300.00 USD, delivery block '01', billing block ''

C. Follow to_Item (what was ordered)
   item 10: 4 PC VALVE-40 (Control valve)
   item 20: 10 PC SEAL-7 (Seal kit)

D. Follow to_Partner (who is involved)
   partner function SP: customer CUST-A
   partner function SH: customer CUST-A

E. Cross into the Business Partner API (the sold-to customer)
   A_Customer CUST-A: Northwind Pumps Inc.; central blocks: order '', delivery '', billing ''
   A_BusinessPartner 1000042: Northwind Pumps Inc., category 2, centrally blocked: False

F. Save the trace and the data dictionary
   Wrote blocked_order_trace.json (2 items, 2 partners)
   Wrote data_dictionary.md (89 lines, 4855 characters)
   First rows of the A_SalesOrder table in the dictionary:
   | SalesOrder | Sales Order | String(10) | Key. |
   | SalesOrderType | Sales Order Type | String(4) | A code defined in configuration. |
   | SalesOrganization | Sales Organization | String(4) | An organizational unit key defined in configuration. |
   | SoldToParty | Sold-to Party | String(10) |  |