oq-ai-usage

Per-user and per-tenant AI token and cost usage recording and analytics, for Python. A small, framework-agnostic core with two optional extras, built so any project can adopt it without pulling in a pricing library or an ORM it does not already use.

Extracted for reuse alongside oq-ai-router (provider-agnostic LLM routing) rather than designed in the abstract: this package answers the question that library deliberately leaves open -- once a call has been routed and answered, who used how much, at what token counts, and at what cost.

What this is, and what it deliberately is not

This package ships no table of its own and no default pricing model. A shared library owning a table in a multi-tenant host is exactly the kind of coupling that breaks the host's own tenancy and row-level-security story; a library inventing a "reasonable default" comparison model for cost-avoided figures would be reporting a number nobody actually decided was correct. Both are refused structurally, not just by convention -- see the honesty rules below.

Install

pip install oq-ai-usage                 # core only: events + analytics
pip install oq-ai-usage[pricing]        # + cost estimation via litellm
pip install oq-ai-usage[sqlalchemy]     # + persistence helpers
pip install "oq-ai-usage[pricing,sqlalchemy]"   # both

The core (UsageEvent, analytics.py) has zero hard dependencies. pricing.py and storage.py both import their extra lazily -- importing this package, or either of those two modules, never requires litellm or sqlalchemy to be installed. Only calling a function that genuinely needs one does, and it raises a named error (PricingUnavailable, or a plain RuntimeError from storage.py) if the extra is missing, rather than degrading silently.

The honesty rules

These are enforced in code, not just documented, and every one of them has a test:

1. No reference model, no cost-avoided figure. A call served by a local model has no cloud invoice of its own. The only honest "what did running this locally save" figure is "what the cloud-equivalent call would have cost", and that number does not exist until the caller names which cloud model is the equivalent. There is no default reference model anywhere in this package.
2. An unknown cost is None, never zero. A model absent from the price map, or no reference model supplied, both mean "not known", not "free". analytics.py and storage.py's SQL aggregates both count such events separately (cost_known_count, cost_unknown_count) and never fold an unknown cost into a sum as if it were 0.0. A group that is genuinely, verifiably free (every event priced at exactly $0.00) is a different, real fact from a group nobody could price at all, and the two render differently: cost_usd=0.0 versus cost_usd=None.
3. The price basis date is surfaced honestly, which today means None. See pricing.py's own docstring for the measured finding: LiteLLM's bundled price map, as actually shipped, carries no per-model "as of" date in its schema. This package still looks for one (in case a future LiteLLM release adds one) rather than hardcoding None outright, but every caller should expect None today.
4. left_the_building is Optional[bool], not bool. None means the host never determined whether the call left the building, which is a different fact from False (determined, and it did not). Nothing in this package defaults a missing value to False.

Quickstart

from datetime import datetime, timezone
from oq_ai_usage import UsageEvent, pricing, analytics

# A call your own inference layer already made and answered.
estimate = pricing.cost_of_call("openai/gpt-4o-mini", prompt_tokens=812, completion_tokens=140)

event = UsageEvent(
    user_id="u_42",
    tenant_id="t_9",
    purpose="tender-extraction",
    provider="openai",
    model="openai/gpt-4o-mini",
    prompt_tokens=812,
    completion_tokens=140,
    lane="openai_compatible_cloud",
    left_the_building=True,
    created_at=datetime.now(timezone.utc),
    cost_usd=estimate.cost_usd,
    price_basis_date=estimate.price_basis_date,
    cost_reason=estimate.reason,
)

by_tenant = analytics.totals_by_tenant([event])

A call served locally has no cost of its own; report what it AVOIDED spending, against a reference model you name explicitly:

avoided = pricing.cloud_equivalent_cost_avoided(
    reference_model="anthropic/claude-3-5-haiku-latest",
    prompt_tokens=812,
    completion_tokens=140,
)
# avoided.cost_usd is None if you omit reference_model. It is never guessed.

FastAPI + SQLAlchemy hosts

Declare your own model on your own Base, splicing in the shared column set (or inheriting the shared mixin) so it has the same shape every adopter's table has, then add whatever tenancy and row-level-security columns and policies your own application already uses:

from sqlalchemy import Integer
from sqlalchemy.orm import DeclarativeBase, mapped_column
from oq_ai_usage import storage

class Base(DeclarativeBase):
    pass

class AiUsageEvent(Base):
    __tablename__ = "ai_usage_events"
    id = mapped_column(Integer, primary_key=True)
    locals().update(storage.usage_event_columns())
    # ... your own tenant_id ForeignKey, RLS policy, indexes, etc.

Then, inside a request handler that already has a Session:

from oq_ai_usage import storage

async def record(session, event: UsageEvent):
    storage.record_event(session, AiUsageEvent, event)
    await session.commit()

totals = storage.query_totals_by_tenant(session, AiUsageEvent)
totals_today = storage.query_totals_by_day(session, AiUsageEvent, tenant_id="t_9")

storage.py asserts nothing about row-level security or tenant isolation -- that is entirely your model and your database's policies. Its own tests run against in-memory SQLite, which has no RLS implementation at all; they prove column shape and aggregation SQL, never a tenancy guarantee.

API surface

• oq_ai_usage.UsageEvent -- the one fact this package works over. Frozen dataclass: user_id, tenant_id, purpose, provider, model, prompt_tokens, completion_tokens, lane, left_the_building, created_at, plus optional cost_usd / price_basis_date / cost_reason.
• oq_ai_usage.pricing -- cost_of_call(), cloud_equivalent_cost_avoided(), CostEstimate, PricingUnavailable. Requires the pricing extra.
• oq_ai_usage.analytics -- totals(), totals_by(), totals_by_user(), totals_by_tenant(), totals_by_purpose(), totals_by_day(), UsageTotals. Pure Python, no dependencies, works over any iterable of UsageEvent.
• oq_ai_usage.storage -- usage_event_columns(), get_usage_event_mixin(), record_event(), query_totals_by_user(), query_totals_by_tenant(), query_totals_by_purpose(), query_totals_by_day(). Requires the sqlalchemy extra.

Testing

pip install -e ".[dev]"
pytest -q

The dev extra installs sqlalchemy (so storage.py is fully tested) and deliberately not litellm, so the suite also proves pricing.PricingUnavailable fires for real when the pricing extra is absent, rather than that path being untested.

Licence

Dual licensed: AGPL v3 or later for community use, plus a proprietary commercial licence for anyone who needs to keep modifications private or embed this in a closed-source product. See LICENSING.md.

Note the Affero clause: running a modified version as a network service obliges you to offer its source to that service's users.

A CLA must be in place before any outside contribution is accepted, because dual licensing only works while the copyright is wholly owned. Until then this repository does not accept contributions.