Tracking events

nozle.track() submits events to base_url/api/v1/events, defaulting to https://api.nozle.app/engine/api/v1/events. Subscription discovery uses events_url/api/v1/subscriptions.

from nozle import Nozle

nozle = Nozle()  # reads NOZLE_SECRET_KEY

nozle.track(
    "workspace_123",
    "api_calls",
    metadata={"model": "example-model", "tokens": 1_500},
    subscription_id="workspace_123_subscription",
    transaction_id="request_0183f",
    timestamp="2026-08-04T10:30:00Z",
)

Signature

nozle.track(
    customer_id,
    event,
    metadata=None,
    subscription_id=None,
    transaction_id=None,
    timestamp=None,
)

Argument	Description
`customer_id`	External Nozle customer ID.
`event`	Feature event code.
`metadata`	JSON-compatible aggregation or attribution properties.
`subscription_id`	Explicit external subscription target.
`transaction_id`	Stable event ID. A UUID is generated when omitted.
`timestamp`	Optional. RFC 3339 with an explicit offset or `Z`, or Unix seconds as a numeric string. Omit it and the server uses the time it received the event.

A naive datetime without a timezone is rejected. datetime.now().isoformat() raises NozleValidationError: track timestamp must be RFC3339 or Unix seconds before any request. Use an aware value, for example datetime.now(timezone.utc).isoformat().

Subscription resolution

When subscription_id is omitted, the SDK queries the Events API for active subscriptions:

one active subscription is cached and reused until the client changes that customer’s subscriptions (see below);
no active subscription raises NozleAPIError (no active subscription for customer '…'); and
multiple active subscriptions raise NozleAPIError (has 2 active subscriptions; specify subscription_id) and require an explicit ID.

From 0.10.0, subscribe(), checkout(), cancel_subscription() and apply_subscription_transition() clear that customer’s cached subscription, even when the call fails, so the next track() looks it up again. Earlier versions never refreshed the cache. checkout_many and the Entity calls do not clear it, because they act on Entity subscriptions. A change made outside this client instance still needs an explicit subscription_id. A second subscribe adds a subscription; it does not replace the first, and with two active subscriptions track() raises until you pass subscription_id.

The lookup errors carry status_code 200. The SDK detects the failure, not an HTTP error. Do not treat 200 as success when you check status_code.

Supply the subscription explicitly for customers with multiple subscriptions and latency-sensitive paths.

Idempotency

Use a deterministic transaction ID and stable event timestamp for each logical event. Reuse both on retries.

from datetime import datetime, timezone

transaction_id = f"generation:{generation_id}:completed"
timestamp = datetime.now(timezone.utc).isoformat()  # store it with the event and reuse it on retries

nozle.track(
    customer_id,
    "text_generation",
    metadata={
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
    },
    subscription_id=subscription_id,
    transaction_id=transaction_id,
    timestamp=timestamp,
)

A timeout, a network error or a 5xx can arrive after Nozle recorded the event. Nozle records an event once per transaction_id, so a retry with the same ID, the same event and the same timestamp does not bill it twice; a new ID is a new event. From nozle-sdk 0.13.3, the error track() raises carries the ID it sent as error.transaction_id (on NozleAPIError, and on NozleTransportError for a network error or a timeout), also when the SDK generated it. Earlier versions lose a generated ID when the call raises, so pass your own transaction_id there. Pass a timestamp and keep it whenever you may retry.

from datetime import datetime, timezone

from nozle import NozleError

timestamp = datetime.now(timezone.utc).isoformat()
try:
    nozle.track(customer_id, "api_call", subscription_id=subscription_id, timestamp=timestamp)
except NozleError as error:
    nozle.track(
        customer_id,
        "api_call",
        subscription_id=subscription_id,
        timestamp=timestamp,
        transaction_id=error.transaction_id,
    )

FastAPI worker pattern

from datetime import timezone

from fastapi import FastAPI

app = FastAPI()


@app.post("/events/generation-completed")
def generation_completed(event: GenerationCompleted) -> dict[str, bool]:
    nozle.track(
        event.customer_id,
        "text_generation",
        metadata={"tokens": event.tokens},
        subscription_id=event.subscription_id,
        transaction_id=f"generation:{event.id}:completed",
        timestamp=event.completed_at.astimezone(timezone.utc).isoformat(),  # timezone-aware, so the SDK accepts it
    )
    return {"accepted": True}

For high-volume workloads, enqueue your business event and call track() from a durable worker. Use usage.track() instead when the operation must atomically enforce product credits.

Setup

Billing

Customers

Entitlements

Margin Intelligence

LLM Usage

Tracking events

Signature

Subscription resolution

Idempotency

FastAPI worker pattern

​Signature

​Subscription resolution

​Idempotency

​FastAPI worker pattern

Signature

Subscription resolution

Idempotency

FastAPI worker pattern