Python reference client

The Python example is a complete pull-first client for Python 3.11 or newer and uses only the standard library. Begin in the synthetic sandbox. Never expose private keys, link codes, cursors, random resource IDs, P12 passwords, or API response bodies in source control or routine logs.

Download the complete Python source.

Requirements and environment#

Place the BOT certificate chain and your locally generated private key in a protected service-secret directory.

export BOT_CERT='/secure/path/bot-integration-certificate-chain.pem'
export BOT_KEY='/secure/path/bot-integration-client-key.pem'
export BOT_KEY_PASSWORD='load-this-from-your-secret-manager'
export BOT_EXPECTED_ENVIRONMENT='sandbox'
python3 bot_integration_client.py

The optional password belongs in the process manager or secret store, never in the repository, image, command line, or a checked-in environment file.

Configure mTLS safely#

The standard-library SSL context validates BOT's public certificate and hostname while presenting the environment-specific client certificate.

context = ssl.create_default_context()
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.load_cert_chain(
    required_environment("BOT_CERT"),
    required_environment("BOT_KEY"),
    password=os.environ.get("BOT_KEY_PASSWORD") or None,
)

The custom redirect handler refuses redirects so client credentials are never presented to another target. Requests are restricted to the exact HTTPS API host, have a timeout, and stop reading after 8 MiB.

Capabilities and production gate#

Call capabilities first and verify every independent gate.

expected_environment = os.environ.get(
    "BOT_EXPECTED_ENVIRONMENT", "sandbox"
)
capabilities = request("GET", "/capabilities")[2]

if capabilities.get("environment") != expected_environment:
    raise RuntimeError("Certificate is bound to the wrong environment.")

if expected_environment == "production" and (
    capabilities.get("integration_status") != "active"
    or capabilities.get("production_data_enabled") is not True
):
    raise RuntimeError("Production is not active/data-enabled.")

Also inspect exact scopes and quota values. A prepared production credential is not proof that the integration or its data access is active.

Redeem a ten-digit link code#

Keep the code as text so a leading zero is not lost.

code = required_environment("BOT_LINK_CODE")
if re.fullmatch(r"[0-9]{10}", code) is None:
    raise RuntimeError("The link code must contain exactly ten digits.")

linked = request(
    "POST",
    "/link-codes/redeem",
    {"code": code},
)[2]

Do not log it and do not blindly retry an uncertain timeout. The response contains only integration-specific random identifiers.

Initial and delta sync#

The synchronize helper performs initial or delta pages with opaque cursors and ETags.

def commit_page(items, next_cursor, next_etag, has_more):
    # In one database transaction:
    # apply every item in order and persist next_cursor + next_etag.
    pass

synchronize(cursor, etag, commit_page)

Commit the page and its cursor together. On failure keep the previous cursor. Treat 304 as unchanged, and never decode, compare, or construct a cursor. A tombstone immediately removes authorization and triggers deletion.

Process the documented fields#

Validate each upsert and persist only the documented allowlisted projection:

projection = project_allowlisted_resource(
    item["resource"]["type"], item["data"]
)

model = projection["model"]
variant = projection["variant"]
delivery_center = projection["delivery_center"]
option_codes = projection["option_codes"]
color = projection["configuration"]["color"]
wheels = projection["configuration"]["wheels"]
interior = projection["configuration"]["interior"]

An order always contains the delivery_center key, although its value can be None. Consume BOT's returned variant and readable configuration. Do not guess from a raw option code when a value remains unknown.

Detail and complete history#

Use a random ID received from redemption or sync:

export BOT_RESOURCE_COLLECTION='orders'
export BOT_RESOURCE_ID='ord_random_integration_specific_id'
python3 bot_integration_client.py

detail = request(
    "GET",
    f"/{collection}/{urllib.parse.quote(resource_id, safe='')}",
)[2]

projection = project_allowlisted_resource(
    "order" if collection == "orders" else "vehicle",
    detail["data"],
)

read_complete_history(collection, resource_id, cursor, etag, commit_page)

History cursors are separate from sync. Preserve sequences and represent the download as incomplete while history_complete is false, even at the final page.

Read aggregate statistics#

The statistics endpoint exposes fixed k-safe aggregate cells only.

statistics = request("GET", "/statistics/orders")[2]

for cell in statistics.get("data", {}).get("orders_total", []):
    country = cell["dimensions"]["country"]
    value = cell["value"]
    print(f"{country}: {value} opted-in orders")

Use the direct orders_total; never sum an incomplete suppressed breakdown. 404 statistics_not_available is a valid state before a safe generation has been published.

Acknowledge deletion#

Delete every controlled copy before acknowledgement.

revocation_id = required_environment("BOT_REVOCATION_ID")
if os.environ.get("BOT_DELETION_COMPLETED") != "1":
    raise RuntimeError("Deletion has not been confirmed locally.")

acknowledge_deletion(revocation_id)

The acknowledgement is idempotent. Respect Retry-After after 429, add local jitter, and never rotate certificates as a rate-limit workaround.

Expected sandbox run#

The maintained fixture validates as Model Y / Long Range AWD with Quicksilver paint, Induction 20-inch wheels, seven option codes, and the clearly synthetic delivery center. Its complete hard-coded expected JSON is rendered below without contacting the API.