Skip to content

Python (httpx) examples — Ludex ingestion

These examples use httpx (pip install httpx). They match the current FastAPI contract: POST /v1/ingest/event and POST /v1/ingest/batch, 202 on success.

Configure from environment variables or your own config layer:

import os
import uuid
from datetime import datetime, timezone

import httpx

BASE_URL = os.environ.get("LUDEX_BASE_URL", "https://ingest.ludexstudio.com").rstrip("/")
API_KEY = os.environ["LUDEX_API_KEY"]
PROJECT_ID = os.environ["LUDEX_PROJECT_ID"]
ENVIRONMENT = os.environ["LUDEX_ENVIRONMENT"]  # must match credential environment_id

Shared client helper

def ludex_headers(
    *,
    idempotency_key: str | None = None,
    correlation_id: str | None = None,
    request_id: str | None = None,
) -> dict[str, str]:
    h: dict[str, str] = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    if idempotency_key:
        h["Idempotency-Key"] = idempotency_key
    if correlation_id:
        h["X-Correlation-Id"] = correlation_id
    if request_id:
        h["X-Request-Id"] = request_id
    return h


def iso_utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

Single event (scope: ingest:events)

def send_single_event(client: httpx.Client) -> httpx.Response:
    body = {
        "project_id": PROJECT_ID,
        "environment": ENVIRONMENT,
        "event": {
            "event_name": "level_complete",
            "timestamp": iso_utc_now(),
            "player_id": "player-uuid-123",
            "session_id": str(uuid.uuid4()),
            "platform": "Windows",
            "properties": {"level": 3, "score": 1200},
            "sdk": {"name": "python-httpx-example", "version": "1.0.0"},
            "event_id": str(uuid.uuid4()),  # stable across retries for the same logical event
        },
    }
    return client.post(
        f"{BASE_URL}/v1/ingest/event",
        json=body,
        headers=ludex_headers(
            idempotency_key="optional-opaque-key",
            correlation_id="corr-001",
        ),
        timeout=30.0,
    )


with httpx.Client() as client:
    r = send_single_event(client)
    r.raise_for_status()
    print(r.status_code, r.json())  # expect 202

Batch (scope: ingest:batch)

def send_batch(client: httpx.Client) -> httpx.Response:
    body = {
        "project_id": PROJECT_ID,
        "environment": ENVIRONMENT,
        "events": [
            {
                "event_name": "match_started",
                "timestamp": iso_utc_now(),
                "session_id": "sess-1",
                "properties": {"mode": "ranked"},
            },
            {
                "event_name": "match_ended",
                "timestamp": iso_utc_now(),
                "session_id": "sess-1",
                "properties": {"duration_sec": 900},
            },
        ],
    }
    return client.post(
        f"{BASE_URL}/v1/ingest/batch",
        json=body,
        headers=ludex_headers(),
        timeout=60.0,
    )


with httpx.Client() as client:
    r = send_batch(client)
    r.raise_for_status()
    print(r.status_code, r.json())  # 202 — "accepted" or "partial"

If the response JSON has "status": "partial", inspect "errors" for per-index failures; still HTTP 202.

Retry with exponential backoff (429 and 5xx)

The server may return 429 (rate limit) or 5xx (e.g. queue failure). Those are generally worth retrying with backoff. 401, 403, 400, 413, and 422 usually require fixing credentials or payload — do not spin forever.

import random
import time


def post_with_retry(
    client: httpx.Client,
    method: str,
    url: str,
    *,
    json: dict,
    headers: dict[str, str],
    max_attempts: int = 5,
    base_delay: float = 0.5,
) -> httpx.Response:
    last: httpx.Response | None = None
    for attempt in range(1, max_attempts + 1):
        r = client.request(method, url, json=json, headers=headers, timeout=60.0)
        last = r
        if r.status_code == 429 or r.status_code >= 500:
            if attempt == max_attempts:
                return r
            sleep_s = base_delay * (2 ** (attempt - 1)) + random.random() * 0.25
            time.sleep(sleep_s)
            continue
        return r
    assert last is not None
    return last