aforo-graphql-metering

Meter every GraphQL operation with AST-accurate complexity scoring — one Aforo event per query/mutation, via a Strawberry extension or an ASGI middleware that works with any GraphQL server.

Version: 1.0.0 · Apache-2.0 · Changelog · User guide

Install

Intended public install:

pip install aforo-graphql-metering
# integrations (pick what you use):
pip install "aforo-graphql-metering[strawberry]"
pip install "aforo-graphql-metering[graphene]"
pip install "aforo-graphql-metering[ariadne]"
pip install "aforo-graphql-metering[httpx]"     # faster HTTP flush than stdlib urllib

Not yet on PyPI — install from source for now:

git clone https://github.com/aforoai/SDKs.git
cd SDKs/aforo-metering-sdks/python-graphql     # folder holding setup.py
pip install -e .
pip install -e ".[strawberry]"            # or [graphene] / [ariadne] / [httpx]

The one required dependency is graphql-core>=3.2 — the SDK parses the operation document itself to score complexity. Without it, record() is a no-op.

Quickstart — Strawberry

Best when you run a Strawberry schema and want per-operation billing without touching resolvers.

import os, strawberry
from aforo_graphql_metering import AforoGraphQlBilling, strawberry_extension

billing = AforoGraphQlBilling(
    tenant_id="tenant_acme",
    product_id="prod_graphql_gateway",
    api_key=os.environ["AFORO_API_KEY"],
    ingestor_url="https://api.aforo.ai",
    schema_version="v2.1",
)

schema = strawberry.Schema(query=Query, extensions=[strawberry_extension(billing)])

Quickstart — ASGI middleware (any GraphQL server)

from starlette.applications import Starlette
from aforo_graphql_metering import AforoGraphQlBilling, asgi_middleware

billing = AforoGraphQlBilling(
    tenant_id="tenant_acme",
    product_id="prod_graphql_gateway",
    api_key=os.environ["AFORO_API_KEY"],
    ingestor_url="https://api.aforo.ai",
)
app = Starlette(routes=[...])
app = asgi_middleware(billing, path="/graphql")(app)   # only intercepts /graphql

Works with Ariadne, graphql-core HTTP, Graphene-ASGI, and custom ASGI GraphQL servers. Each metered operation produces one graphql_api.operations event POSTed to https://api.aforo.ai/v1/ingest/batch with X-API-Key: <api_key> and X-Tenant-Id: <tenant_id>.

⚠ Events are sent to the ingestor's /v1/ingest/batch path as {"events": [...]}, at most 1000 events per request (larger buffers are split). Set ingestor_url to the host only — the SDK appends the path. Use https://api.aforo.ai.

tenant_id is fixed from your config and sent as a header — never read from a caller-controlled value. The default customer-ID extractor reads x-customer-id from request headers (or the Strawberry context); operations with no resolvable customer ID are not metered.

Configuration

Constructor arguments for AforoGraphQlBilling(...):

Retry is fixed at 3 attempts (1s / 2s backoff between them); 408 and 5xx are retried, 429 waits for Retry-After (capped at 60 s), and any other 4xx is not retried.

Walk me through it

Install → wire the extension → run a query → confirm the event in Aforo, step by step, is in USER_GUIDE.md.

What this doesn't cover

It meters operations, not individual resolver fields or DataLoader batches — one event per top-level operation, with complexity/field counts as attributes. It doesn't enforce a complexity budget or reject expensive queries; that's a gateway/server concern. GraphQL subscriptions over WebSocket aren't covered here — use aforo-ws-metering. Pricing and metric mapping are in the Aforo console.