API reference¶
Generated from the apogee-ai-cost source with mkdocstrings. Every symbol below is exported from apogee_ai_cost, so it is part of the supported public surface.
Application · DTOs¶
BenchDTO
dataclass
¶
BudgetDTO
dataclass
¶
CountDTO
dataclass
¶
EstimateDTO
dataclass
¶
RecordDTO
dataclass
¶
RecordDTO(model: str, amount_usd: float, tenant_id: str | None = None, prompt_tokens: int = 0, completion_tokens: int = 0)
ReportDTO
dataclass
¶
Application · Use cases¶
BenchPricingUseCase
¶
Synthetic pricing-loop benchmark across the builtin table.
execute
async
¶
Source code in apogee_ai_cost/application/use_cases/bench_pricing_use_case.py
async def execute(self, calls: int) -> dict[str, float]:
if calls <= 0:
raise ValueError("calls must be positive")
table = PricingTable.builtin()
use_case = EstimateCostUseCase(table)
models = [e.model for e in table.list() if e.prompt_per_1k_usd > 0]
if not models:
raise RuntimeError("no priced models in builtin table")
start = time.perf_counter()
for i in range(calls):
await use_case.execute(TokenUsage(
model=models[i % len(models)],
prompt_tokens=1000,
completion_tokens=200,
))
elapsed = (time.perf_counter() - start) * 1000.0
return {
"calls": float(calls),
"elapsed_ms": elapsed,
"calls_per_second": (calls / elapsed * 1000.0) if elapsed > 0 else 0.0,
}
CheckBudgetUseCase
¶
Source code in apogee_ai_cost/application/use_cases/check_budget_use_case.py
CountTokensUseCase
¶
Source code in apogee_ai_cost/application/use_cases/count_tokens_use_case.py
execute
async
¶
EstimateCostUseCase
¶
Source code in apogee_ai_cost/application/use_cases/estimate_cost_use_case.py
execute
async
¶
execute(usage: TokenUsage) -> CostEstimate
Source code in apogee_ai_cost/application/use_cases/estimate_cost_use_case.py
async def execute(self, usage: TokenUsage) -> CostEstimate:
entry = self._table.get(usage.model)
prompt_cost = (usage.prompt_tokens / 1000.0) * entry.prompt_per_1k_usd
completion_cost = (usage.completion_tokens / 1000.0) * entry.completion_per_1k_usd
embedding_cost = (usage.embedding_tokens / 1000.0) * entry.embedding_per_1k_usd
cache_read_cost = (usage.cache_read_tokens / 1000.0) * entry.cache_read_per_1k_usd
cache_write_cost = (usage.cache_write_tokens / 1000.0) * entry.cache_write_per_1k_usd
total = (
prompt_cost
+ completion_cost
+ embedding_cost
+ cache_read_cost
+ cache_write_cost
)
return CostEstimate(
model=usage.model,
amount_usd=total,
breakdown={
"prompt": prompt_cost,
"completion": completion_cost,
"embedding": embedding_cost,
"cache_read": cache_read_cost,
"cache_write": cache_write_cost,
},
)
GetReportUseCase
¶
Source code in apogee_ai_cost/application/use_cases/get_report_use_case.py
execute
async
¶
Source code in apogee_ai_cost/application/use_cases/get_report_use_case.py
async def execute(self, tenant_id: str | None = None, fmt: str = "json") -> str:
entries = await self._repo.list_by_tenant(tenant_id)
report = CostReport(entries)
if fmt == "json":
return JsonReportFormatter().format(report)
if fmt == "markdown":
return MarkdownReportFormatter().format(report)
raise ValueError(f"unknown report format: {fmt!r}")
RecordUsageUseCase
¶
Records token usage: estimates cost, persists entry, updates budget.
Source code in apogee_ai_cost/application/use_cases/record_usage_use_case.py
execute
async
¶
execute(usage: TokenUsage) -> CostEntry
Source code in apogee_ai_cost/application/use_cases/record_usage_use_case.py
async def execute(self, usage: TokenUsage) -> CostEntry:
estimate = await self._estimate.execute(usage)
entry = CostEntry(
model=usage.model,
tenant_id=usage.tenant_id,
amount_usd=estimate.amount_usd,
prompt_tokens=usage.prompt_tokens,
completion_tokens=usage.completion_tokens,
metadata={
"embedding_tokens": str(usage.embedding_tokens),
},
)
await self._repo.add(entry)
if self._tracker is not None and usage.tenant_id is not None:
await self._tracker.record_spend(usage.tenant_id, estimate.amount_usd)
return entry
Domain¶
Budget
dataclass
¶
Budget(tenant_id: str, limit_usd: float, spent_usd: float = 0.0, currency: Currency = USD, alert_at_pct: float = 0.8)
CostEntry
dataclass
¶
CostEntry(model: str, tenant_id: str | None = None, amount_usd: float = 0.0, currency: Currency = USD, prompt_tokens: int = 0, completion_tokens: int = 0, timestamp_s: float = time(), id: str = (lambda: f'e-{hex[:8]}')(), metadata: dict[str, str] = dict())
CostEstimate
dataclass
¶
CostEstimate(model: str, amount_usd: float = 0.0, currency: Currency = USD, breakdown: dict[str, float] = dict())
breakdown
class-attribute
instance-attribute
¶
Currency
¶
Money
dataclass
¶
PricingEntry
dataclass
¶
PricingEntry(model: str, provider: Provider = UNKNOWN, prompt_per_1k_usd: float = 0.0, completion_per_1k_usd: float = 0.0, embedding_per_1k_usd: float = 0.0, cache_read_per_1k_usd: float = 0.0, cache_write_per_1k_usd: float = 0.0)
Pricing for one model. Rates expressed in USD per 1k tokens.
cache_write_per_1k_usd
class-attribute
instance-attribute
¶
Provider
¶
Bases: str, Enum
TokenUsage
dataclass
¶
TokenUsage(model: str, prompt_tokens: int = 0, completion_tokens: int = 0, embedding_tokens: int = 0, cache_read_tokens: int = 0, cache_write_tokens: int = 0, tenant_id: str | None = None)
Domain · Enums¶
TokenKind
¶
Bases: str, Enum
Domain · Exceptions¶
BudgetExceededException
¶
Bases: CostError
Source code in apogee_ai_cost/domain/exceptions/cost_exceptions.py
CostError
¶
Bases: Exception
Base for apogee-ai-cost errors.
PricingNotFoundException
¶
Domain · Protocols (ports)¶
IBudgetTracker
¶
ICostRepository
¶
IPricingTable
¶
Bases: Protocol
get
¶
get(model: str) -> PricingEntry
list
¶
list() -> list[PricingEntry]
ITokenCounter
¶
Infrastructure¶
CostReport
¶
CostReport(entries: Iterable[CostEntry])
Aggregates a list of CostEntry into a tenant/model summary.
Source code in apogee_ai_cost/infrastructure/reports/cost_report.py
summarise
¶
Source code in apogee_ai_cost/infrastructure/reports/cost_report.py
def summarise(self) -> dict:
by_tenant: dict[str, float] = defaultdict(float)
by_model: dict[str, float] = defaultdict(float)
for e in self._entries:
by_tenant[e.tenant_id or "<anonymous>"] += e.amount_usd
by_model[e.model] += e.amount_usd
return {
"total_usd": sum(e.amount_usd for e in self._entries),
"entries": len(self._entries),
"by_tenant": dict(by_tenant),
"by_model": dict(by_model),
}
InMemoryBudgetTracker
¶
Source code in apogee_ai_cost/infrastructure/budgets/in_memory_budget_tracker.py
set_budget
async
¶
set_budget(tenant_id: str, limit_usd: float, alert_at_pct: float = 0.8) -> Budget
Source code in apogee_ai_cost/infrastructure/budgets/in_memory_budget_tracker.py
async def set_budget(
self, tenant_id: str, limit_usd: float, alert_at_pct: float = 0.8
) -> Budget:
existing = self._budgets.get(tenant_id)
spent = existing.spent_usd if existing else 0.0
budget = Budget(
tenant_id=tenant_id,
limit_usd=limit_usd,
spent_usd=spent,
alert_at_pct=alert_at_pct,
)
self._budgets[tenant_id] = budget
return budget
record_spend
async
¶
record_spend(tenant_id: str, amount_usd: float) -> Budget
Source code in apogee_ai_cost/infrastructure/budgets/in_memory_budget_tracker.py
async def record_spend(self, tenant_id: str, amount_usd: float) -> Budget:
if amount_usd < 0:
raise ValueError("amount_usd must be non-negative")
if tenant_id not in self._budgets:
# Untracked tenants accumulate without limit until set_budget.
self._budgets[tenant_id] = Budget(
tenant_id=tenant_id, limit_usd=float("inf"), spent_usd=amount_usd,
)
return self._budgets[tenant_id]
b = self._budgets[tenant_id]
new_spent = b.spent_usd + amount_usd
updated = replace(b, spent_usd=new_spent)
self._budgets[tenant_id] = updated
if updated.exceeded and self._raise_on_exceed and updated.limit_usd != float("inf"):
raise BudgetExceededException(
tenant=tenant_id,
limit_usd=updated.limit_usd,
spent_usd=updated.spent_usd,
)
return updated
InMemoryCostRepository
¶
Source code in apogee_ai_cost/infrastructure/registries/in_memory_cost_repository.py
JsonPricingLoader
¶
Loads a pricing table from a JSON file with a list of entries.
load
¶
load(path: str | Path) -> PricingTable
Source code in apogee_ai_cost/infrastructure/pricing/json_pricing_loader.py
def load(self, path: str | Path) -> PricingTable:
path = Path(path)
payload = json.loads(path.read_text(encoding="utf-8"))
entries: list[PricingEntry] = []
for item in payload:
entries.append(
PricingEntry(
model=item["model"],
provider=Provider(item.get("provider", "unknown")),
prompt_per_1k_usd=float(item.get("prompt_per_1k_usd", 0.0)),
completion_per_1k_usd=float(item.get("completion_per_1k_usd", 0.0)),
embedding_per_1k_usd=float(item.get("embedding_per_1k_usd", 0.0)),
cache_read_per_1k_usd=float(item.get("cache_read_per_1k_usd", 0.0)),
cache_write_per_1k_usd=float(item.get("cache_write_per_1k_usd", 0.0)),
)
)
return PricingTable(entries)
JsonReportFormatter
¶
format
¶
format(report: CostReport) -> str
MarkdownReportFormatter
¶
format
¶
format(report: CostReport) -> str
Source code in apogee_ai_cost/infrastructure/reports/cost_report.py
def format(self, report: CostReport) -> str:
s = report.summarise()
lines = ["# Cost Report", ""]
lines.append(f"- **Total**: ${s['total_usd']:.6f}")
lines.append(f"- **Entries**: {s['entries']}")
lines.append("")
lines.append("## By tenant")
lines.append("")
for tenant, amount in sorted(s["by_tenant"].items(), key=lambda kv: -kv[1]):
lines.append(f"- `{tenant}`: ${amount:.6f}")
lines.append("")
lines.append("## By model")
lines.append("")
for model, amount in sorted(s["by_model"].items(), key=lambda kv: -kv[1]):
lines.append(f"- `{model}`: ${amount:.6f}")
return "\n".join(lines)
PricingTable
¶
PricingTable(entries: Iterable[PricingEntry] = ())
Source code in apogee_ai_cost/infrastructure/pricing/pricing_table.py
builtin
classmethod
¶
register
¶
register(entry: PricingEntry) -> None
get
¶
get(model: str) -> PricingEntry
list
¶
list() -> list[PricingEntry]
TiktokenCounter
¶
Lazy tiktoken adapter — install via extras=tiktoken.
Source code in apogee_ai_cost/infrastructure/counters/tiktoken_counter.py
count
async
¶
WhitespaceTokenCounter
¶
CI-safe heuristic. Splits on whitespace and applies a 0.75 BPE adjustment.
Real tokenizers (tiktoken) cost network/dependency. This is good enough for budgeting estimates within ~10-20%.
count
async
¶
builtin_entries
¶
builtin_entries() -> tuple[PricingEntry, ...]