Skip to content

symfonic.core.observability.budget_models

budget_models

Shared data types + protocols for the budget tracker.

Split from :mod:symfonic.core.observability.budget to keep each module under the 300-LOC cap. The concrete :class:TokenBudgetTracker lives next door and imports these types.

BudgetCheckResult dataclass 

BudgetCheckResult(
    allowed: bool,
    reason: str | None = None,
    daily_used_usd: float = 0.0,
    daily_limit_usd: float | None = None,
    monthly_used_usd: float = 0.0,
    monthly_limit_usd: float | None = None,
)

Result of a pre-call budget check.

BudgetStore

Bases: Protocol

Optional persistence layer for daily aggregates.

Implementations should be idempotent on conflict — the tracker calls upsert_daily after every LLM call, so the underlying storage must handle UPSERT semantics (ON CONFLICT ... DO UPDATE).

The optional read_today / read_month methods hydrate the tracker's in-memory buckets on demand so that multiple app replicas converge on the same view of a tenant's cumulative usage. A store that only implements upsert_daily still works (writes only) — multi-replica correctness requires all three.

read_month async 

read_month(
    tenant_id: str, year_month: str
) -> BudgetUsage | None

Return the aggregated month-to-date usage for tenant_id.

year_month is the YYYY-MM key produced by :func:month_key. Optional — see :meth:read_today.

Source code in src/symfonic/core/observability/budget_models.py 
 94
 95
 96
 97
 98
 99
100
101
102
async def read_month(  # pragma: no cover
    self, tenant_id: str, year_month: str,
) -> BudgetUsage | None:
    """Return the aggregated month-to-date usage for ``tenant_id``.

    ``year_month`` is the ``YYYY-MM`` key produced by
    :func:`month_key`.  Optional — see :meth:`read_today`.
    """
    ...

read_today async 

read_today(tenant_id: str) -> BudgetUsage | None

Return today's usage row for tenant_id (None if absent).

Optional — implementations without this method get write-through semantics only (single-process correctness).

Source code in src/symfonic/core/observability/budget_models.py 
84
85
86
87
88
89
90
91
92
async def read_today(  # pragma: no cover
    self, tenant_id: str,
) -> BudgetUsage | None:
    """Return today's usage row for ``tenant_id`` (None if absent).

    Optional — implementations without this method get write-through
    semantics only (single-process correctness).
    """
    ...

BudgetUsage dataclass 

BudgetUsage(
    tenant_id: str,
    window: str,
    window_key: str,
    input_tokens: int = 0,
    output_tokens: int = 0,
    cached_tokens: int = 0,
    cache_write_tokens: int = 0,
    cost_usd: float = 0.0,
    request_count: int = 0,
    updated_at: datetime = (lambda: datetime.now(UTC))(),
)

Cumulative usage for a single (tenant, window) pair.

copy_usage 

copy_usage(src: BudgetUsage) -> BudgetUsage

Snapshot a bucket so callers can't mutate tracker state.

Source code in src/symfonic/core/observability/budget_models.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def copy_usage(src: BudgetUsage) -> BudgetUsage:
    """Snapshot a bucket so callers can't mutate tracker state."""
    return BudgetUsage(
        tenant_id=src.tenant_id,
        window=src.window,
        window_key=src.window_key,
        input_tokens=src.input_tokens,
        output_tokens=src.output_tokens,
        cached_tokens=src.cached_tokens,
        cache_write_tokens=src.cache_write_tokens,
        cost_usd=src.cost_usd,
        request_count=src.request_count,
        updated_at=src.updated_at,
    )

month_key 

month_key() -> str

Return the current month key (YYYY-MM).

Source code in src/symfonic/core/observability/budget_models.py 
110
111
112
113
def month_key() -> str:
    """Return the current month key (``YYYY-MM``)."""
    today = date.today()
    return f"{today.year:04d}-{today.month:02d}"

today_key 

today_key() -> str

Return today's ISO date string — used to key daily buckets.

Source code in src/symfonic/core/observability/budget_models.py 
105
106
107
def today_key() -> str:
    """Return today's ISO date string — used to key daily buckets."""
    return date.today().isoformat()