FastAPI and Django reference implementations

Use one validated event model and one delivery client across Python services.

FastAPI public endpoint → Pydantic allowlist → delivery
Django business commit → durable job → worker → GraphJSON

Configuration#

import os

GRAPHJSON_API_KEY = os.environ["GRAPHJSON_API_KEY"]
GRAPHJSON_COLLECTION = os.getenv(
    "GRAPHJSON_COLLECTION",
    "product_events",
)

Let missing required configuration fail startup. Never send the key to a template, browser bundle, mobile client, or exception tracker.

Event contract#

from typing import Literal
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field

class PublicEvent(BaseModel):
    model_config = ConfigDict(extra="forbid")

    event: Literal["pricing_viewed", "signup_started"]
    anonymous_id: UUID
    occurred_at: int = Field(ge=0)
    path: Literal["/pricing", "/signup"]

Keep authenticated user_id and account_id out of the public model. Derive them from server authentication.

Shared asynchronous client#

import json

import httpx

class GraphJSONClient:
    def __init__(self, api_key: str, collection: str):
        self.api_key = api_key
        self.collection = collection
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(5.0),
        )

    async def log(
        self,
        occurred_at: int,
        payload: dict,
    ) -> None:
        response = await self.client.post(
            "https://api.graphjson.com/api/log",
            json={
                "api_key": self.api_key,
                "collection": self.collection,
                "timestamp": occurred_at,
                "json": json.dumps(
                    payload,
                    separators=(",", ":"),
                ),
            },
        )
        response.raise_for_status()

    async def close(self) -> None:
        await self.client.aclose()

Reuse the HTTP client. Creating a new connection pool for every event wastes resources.

The caller owns retry policy. raise_for_status() alone does not distinguish a non-retryable 400 from a retryable 429 or 500.

FastAPI first-party endpoint#

from contextlib import asynccontextmanager
from uuid import uuid4
from fastapi import FastAPI, Request, Response

graphjson = GraphJSONClient(
    GRAPHJSON_API_KEY,
    GRAPHJSON_COLLECTION,
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await graphjson.close()

app = FastAPI(lifespan=lifespan)

@app.post("/analytics", status_code=202)
async def collect(
    event: PublicEvent,
    request: Request,
) -> Response:
    session = await optional_session(request)
    payload = {
        **event.model_dump(mode="json"),
        "event_id": str(uuid4()),
        "user_id": session.user_id if session else None,
        "account_id": (
            session.active_account_id if session else None
        ),
        "environment": os.getenv(
            "APP_ENV",
            "development",
        ),
        "schema_version": 1,
    }

    try:
        await graphjson.log(event.occurred_at, payload)
    except Exception as error:
        logger.warning(
            "analytics_delivery_failed",
            extra={
                "event": event.event,
                "event_id": payload["event_id"],
                "reason": type(error).__name__,
            },
        )

    return Response(status_code=202)

Add request-size, consent, origin, and abuse controls appropriate to the application.

FastAPI background tasks run in the web process and are not a durable queue. Use them only for events you can lose. Use a database job or message broker for facts that must reconcile.

Django transaction boundary#

Create the business state and durable job together:

from django.db import transaction
from django.utils import timezone

with transaction.atomic():
    order = Order.objects.create(**approved_order)

    AnalyticsOutbox.objects.create(
        event_id=f"order:{order.id}:completed",
        occurred_at=int(
            order.completed_at.timestamp()
        ),
        collection="order_events",
        payload={
            "event": "order_completed",
            "event_id": f"order:{order.id}:completed",
            "account_id": str(order.account_id),
            "order_id": str(order.id),
            "amount_minor": order.amount_minor,
            "currency": order.currency,
            "schema_version": 1,
        },
    )

Do not perform a remote GraphJSON request inside the database transaction.

If you enqueue a task rather than storing an outbox, use transaction.on_commit so a worker cannot run before the business transaction commits:

transaction.on_commit(
    lambda: deliver_analytics.delay(outbox.id)
)

The durable outbox remains necessary if publishing the task can fail after the commit.

Bulk worker#

async def send_batch(rows: list[AnalyticsOutbox]) -> None:
    collection = rows[0].collection
    if any(row.collection != collection for row in rows):
        raise ValueError("mixed_collection_batch")

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            "https://api.graphjson.com/api/bulk-log",
            json={
                "api_key": GRAPHJSON_API_KEY,
                "collection": collection,
                "timestamps": [
                    row.occurred_at for row in rows
                ],
                "jsons": [
                    json.dumps(
                        row.payload,
                        separators=(",", ":"),
                    )
                    for row in rows
                ],
            },
        )

    if response.status_code == 400:
        raise InvalidAnalyticsBatch(response.text[:300])
    if response.status_code in {429, 500}:
        raise RetryableAnalyticsBatch(
            response.status_code,
        )
    response.raise_for_status()

Claim no more than 50 compatible rows. Use database locking or a lease so two workers cannot own the same row concurrently.

Retries preserve payload, event ID, and occurrence time.

Safe logging#

Log:

event name
event ID
collection
status class
attempt
safe reason code

Do not log:

API key
complete payload
authorization headers
personal fields
full response when it can contain data

Tests#

FastAPI:

rejects extra properties
derives tenant ID from the session
returns 202 when disposable analytics fails
enforces consent and rate policy

Django:

outbox rolls back with the business transaction
task publishes only after commit
worker batches one collection
400 quarantines
429, timeout, and 500 retry
event ID remains stable

Use dependency injection or an HTTP mock for most tests. Keep one bounded live contract test in a non-production workspace.

Deployment checklist#

Secret is environment-only.
Pydantic forbids unexpected public fields.
HTTP client has explicit timeout.
Shared clients close during application shutdown.
Disposable analytics cannot break the request.
Important facts commit to a durable outbox.
Django remote calls stay outside transactions.
Worker claims are exclusive and bounded.
Retry classification uses HTTP status.
Logs contain no payloads or secrets.

Continue with Executable event contracts, Reliable event delivery, and Bulk logging.