Saltar a contenido

API reference

Generated from the apogee-ai-billing source with mkdocstrings. Every symbol below is exported from apogee_ai_billing, so it is part of the supported public surface.

Application · DTOs

BenchDTO dataclass

Python
BenchDTO(tenants: int = 100)

tenants class-attribute instance-attribute

Python
tenants: int = 100

InvoiceDTO dataclass

Python
InvoiceDTO(tenant_id: str, period_start_s: float, period_end_s: float | None = None, format: str = 'json')

tenant_id instance-attribute

Python
tenant_id: str

period_start_s instance-attribute

Python
period_start_s: float

period_end_s class-attribute instance-attribute

Python
period_end_s: float | None = None

format class-attribute instance-attribute

Python
format: str = 'json'

RecordDTO dataclass

Python
RecordDTO(tenant_id: str, metric: str = 'tokens', quantity: float = 0.0)

tenant_id instance-attribute

Python
tenant_id: str

metric class-attribute instance-attribute

Python
metric: str = 'tokens'

quantity class-attribute instance-attribute

Python
quantity: float = 0.0

SubscribeDTO dataclass

Python
SubscribeDTO(tenant_id: str, plan_code: str)

tenant_id instance-attribute

Python
tenant_id: str

plan_code instance-attribute

Python
plan_code: str

Application · Use cases

BenchInvoicingUseCase

Synthetic load: build N invoices with mixed plans + record per tenant.

execute async

Python
execute(tenants: int) -> dict[str, float]
Source code in apogee_ai_billing/application/use_cases/bench_invoicing_use_case.py
Python
async def execute(self, tenants: int) -> dict[str, float]:
    if tenants <= 0:
        raise ValueError("tenants must be positive")
    flat = FlatPlan(
        code="flat", name="Flat",
        monthly_usd=49.0, included_tokens=100_000,
        overage_per_1k_usd=0.01,
    )
    tiered = TieredPlan(
        code="tiered", name="Tiered",
        monthly_usd=0.0,
        tiers=(
            PriceTier(up_to=10_000, price_per_1k_usd=0.02),
            PriceTier(up_to=100_000, price_per_1k_usd=0.015),
            PriceTier(up_to=-1, price_per_1k_usd=0.005),
        ),
    )
    meter = TokenMeter()
    repo = InMemoryInvoiceRepository()
    builder = InvoiceBuilder()
    period_start = time.time() - 86400.0

    start = time.perf_counter()
    for i in range(tenants):
        tenant = f"t-{i:05d}"
        await meter.record_tokens(tenant, 50_000 + i * 10)
        plan = flat if i % 2 == 0 else tiered
        usage = await meter.aggregate(tenant, "tokens", since_s=period_start)
        invoice = builder.build(
            tenant_id=tenant,
            plan=plan,
            usage_units=usage,
            period_start_s=period_start,
        )
        await repo.save(invoice)
    elapsed = (time.perf_counter() - start) * 1000.0
    return {
        "tenants": float(tenants),
        "invoices": float(await repo.count()),
        "elapsed_ms": elapsed,
        "invoices_per_second": (tenants / elapsed * 1000.0) if elapsed > 0 else 0.0,
    }

GenerateInvoiceUseCase

Python
GenerateInvoiceUseCase(plan_repo, subscription_repo, meter, invoice_repo, builder: InvoiceBuilder | None = None)
Source code in apogee_ai_billing/application/use_cases/generate_invoice_use_case.py
Python
def __init__(
    self,
    plan_repo,
    subscription_repo,
    meter,
    invoice_repo,
    builder: InvoiceBuilder | None = None,
) -> None:
    self._plans = plan_repo
    self._subs = subscription_repo
    self._meter = meter
    self._invoices = invoice_repo
    self._builder = builder or InvoiceBuilder()

execute async

Python
execute(tenant_id: str, period_start_s: float, period_end_s: float | None = None, metric: str = 'tokens') -> Invoice
Source code in apogee_ai_billing/application/use_cases/generate_invoice_use_case.py
Python
async def execute(
    self,
    tenant_id: str,
    period_start_s: float,
    period_end_s: float | None = None,
    metric: str = "tokens",
) -> Invoice:
    subscription = await self._subs.get(tenant_id)
    plan = self._plans.get(subscription.plan_code)
    end = period_end_s if period_end_s is not None else time.time()
    usage = await self._meter.aggregate(
        tenant_id, metric, since_s=period_start_s
    )
    invoice = self._builder.build(
        tenant_id=tenant_id,
        plan=plan,
        usage_units=usage,
        period_start_s=period_start_s,
        period_end_s=end,
    )
    await self._invoices.save(invoice)
    return invoice

ListPlansUseCase

Python
ListPlansUseCase(plan_repo)
Source code in apogee_ai_billing/application/use_cases/list_plans_use_case.py
Python
def __init__(self, plan_repo) -> None:
    self._plans = plan_repo

execute async

Python
execute() -> list
Source code in apogee_ai_billing/application/use_cases/list_plans_use_case.py
Python
async def execute(self) -> list:
    return list(self._plans.list())

RecordUsageUseCase

Python
RecordUsageUseCase(meter)
Source code in apogee_ai_billing/application/use_cases/record_usage_use_case.py
Python
def __init__(self, meter) -> None:
    self._meter = meter

execute async

Python
execute(tenant_id: str, metric: str, quantity: float, **metadata: str) -> UsageRecord
Source code in apogee_ai_billing/application/use_cases/record_usage_use_case.py
Python
async def execute(
    self, tenant_id: str, metric: str, quantity: float, **metadata: str
) -> UsageRecord:
    record = UsageRecord(
        tenant_id=tenant_id,
        metric=metric,
        quantity=quantity,
        metadata={k: str(v) for k, v in metadata.items()},
    )
    await self._meter.record(record)
    return record

SubscribePlanUseCase

Python
SubscribePlanUseCase(plan_repo, subscription_repo)
Source code in apogee_ai_billing/application/use_cases/subscribe_plan_use_case.py
Python
def __init__(self, plan_repo, subscription_repo) -> None:
    self._plans = plan_repo
    self._subs = subscription_repo

execute async

Python
execute(tenant_id: str, plan_code: str) -> Subscription
Source code in apogee_ai_billing/application/use_cases/subscribe_plan_use_case.py
Python
async def execute(self, tenant_id: str, plan_code: str) -> Subscription:
    self._plans.get(plan_code)  # raises if missing
    subscription = Subscription(tenant_id=tenant_id, plan_code=plan_code)
    await self._subs.upsert(subscription)
    return subscription

Domain

BillingPeriod

Bases: str, Enum

DAILY class-attribute instance-attribute

Python
DAILY = 'daily'

MONTHLY class-attribute instance-attribute

Python
MONTHLY = 'monthly'

QUARTERLY class-attribute instance-attribute

Python
QUARTERLY = 'quarterly'

YEARLY class-attribute instance-attribute

Python
YEARLY = 'yearly'

Invoice dataclass

Python
Invoice(tenant_id: str, plan_code: str, period_start_s: float, period_end_s: float, lines: tuple[InvoiceLine, ...] = (), status: InvoiceStatus = DRAFT, id: str = (lambda: f'inv-{hex[:10]}')(), issued_at_s: float = time(), metadata: dict[str, str] = dict())

tenant_id instance-attribute

Python
tenant_id: str

plan_code instance-attribute

Python
plan_code: str

period_start_s instance-attribute

Python
period_start_s: float

period_end_s instance-attribute

Python
period_end_s: float

lines class-attribute instance-attribute

Python
lines: tuple[InvoiceLine, ...] = ()

status class-attribute instance-attribute

Python
status: InvoiceStatus = DRAFT

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'inv-{hex[:10]}')

issued_at_s class-attribute instance-attribute

Python
issued_at_s: float = field(default_factory=time)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

total_usd property

Python
total_usd: float

line_count property

Python
line_count: int

InvoiceLine dataclass

Python
InvoiceLine(description: str, quantity: float, unit_price_usd: float, amount_usd: float)

description instance-attribute

Python
description: str

quantity instance-attribute

Python
quantity: float

unit_price_usd instance-attribute

Python
unit_price_usd: float

amount_usd instance-attribute

Python
amount_usd: float

InvoiceStatus

Bases: str, Enum

DRAFT class-attribute instance-attribute

Python
DRAFT = 'draft'

OPEN class-attribute instance-attribute

Python
OPEN = 'open'

PAID class-attribute instance-attribute

Python
PAID = 'paid'

VOID class-attribute instance-attribute

Python
VOID = 'void'

UNCOLLECTIBLE class-attribute instance-attribute

Python
UNCOLLECTIBLE = 'uncollectible'

Plan dataclass

Python
Plan(code: str, name: str, kind: PlanKind = FLAT, period: BillingPeriod = MONTHLY, base_fee_usd: float = 0.0, included_units: int = 0, overage_price_per_1k_usd: float = 0.0, tiers: tuple[PriceTier, ...] = (), metadata: dict[str, str] = dict())

code instance-attribute

Python
code: str

name instance-attribute

Python
name: str

kind class-attribute instance-attribute

Python
kind: PlanKind = FLAT

period class-attribute instance-attribute

Python
period: BillingPeriod = MONTHLY

base_fee_usd class-attribute instance-attribute

Python
base_fee_usd: float = 0.0

included_units class-attribute instance-attribute

Python
included_units: int = 0

overage_price_per_1k_usd class-attribute instance-attribute

Python
overage_price_per_1k_usd: float = 0.0

tiers class-attribute instance-attribute

Python
tiers: tuple[PriceTier, ...] = ()

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

PriceTier dataclass

Python
PriceTier(up_to: int, price_per_1k_usd: float)

A pricing tier defined by an upper-bound usage volume.

The tier price applies to consumption that falls within the band (previous_upper, up_to].

up_to instance-attribute

Python
up_to: int

price_per_1k_usd instance-attribute

Python
price_per_1k_usd: float

Subscription dataclass

Python
Subscription(tenant_id: str, plan_code: str, started_at_s: float = time(), active: bool = True, metadata: dict[str, str] = dict())

tenant_id instance-attribute

Python
tenant_id: str

plan_code instance-attribute

Python
plan_code: str

started_at_s class-attribute instance-attribute

Python
started_at_s: float = field(default_factory=time)

active class-attribute instance-attribute

Python
active: bool = True

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

UsageRecord dataclass

Python
UsageRecord(tenant_id: str, metric: str, quantity: float, timestamp_s: float = time(), id: str = (lambda: f'u-{hex[:8]}')(), metadata: dict[str, str] = dict())

tenant_id instance-attribute

Python
tenant_id: str

metric instance-attribute

Python
metric: str

quantity instance-attribute

Python
quantity: float

timestamp_s class-attribute instance-attribute

Python
timestamp_s: float = field(default_factory=time)

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'u-{hex[:8]}')

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

Domain · Enums

PlanKind

Bases: str, Enum

FLAT class-attribute instance-attribute

Python
FLAT = 'flat'

TIERED class-attribute instance-attribute

Python
TIERED = 'tiered'

USAGE_BASED class-attribute instance-attribute

Python
USAGE_BASED = 'usage_based'

Domain · Exceptions

BillingError

Bases: Exception

Base for apogee-ai-billing errors.

InvoiceNotFoundException

Bases: BillingError

PlanNotFoundException

Python
PlanNotFoundException(code: str)

Bases: BillingError

Source code in apogee_ai_billing/domain/exceptions/billing_exceptions.py
Python
def __init__(self, code: str) -> None:
    super().__init__(f"Plan not found: {code!r}")
    self.code = code

code instance-attribute

Python
code = code

SubscriptionNotFoundException

Python
SubscriptionNotFoundException(tenant_id: str)

Bases: BillingError

Source code in apogee_ai_billing/domain/exceptions/billing_exceptions.py
Python
def __init__(self, tenant_id: str) -> None:
    super().__init__(f"No subscription for tenant {tenant_id!r}")
    self.tenant_id = tenant_id

tenant_id instance-attribute

Python
tenant_id = tenant_id

Domain · Protocols (ports)

IInvoiceRepository

Bases: Protocol

save async

Python
save(invoice: Invoice) -> None
Source code in apogee_ai_billing/domain/services/i_invoice_repository.py
Python
async def save(self, invoice: Invoice) -> None: ...

get async

Python
get(invoice_id: str) -> Invoice
Source code in apogee_ai_billing/domain/services/i_invoice_repository.py
Python
async def get(self, invoice_id: str) -> Invoice: ...

list_by_tenant async

Python
list_by_tenant(tenant_id: str) -> Iterable[Invoice]
Source code in apogee_ai_billing/domain/services/i_invoice_repository.py
Python
async def list_by_tenant(self, tenant_id: str) -> Iterable[Invoice]: ...

IMeter

Bases: Protocol

name instance-attribute

Python
name: str

record async

Python
record(record: UsageRecord) -> None
Source code in apogee_ai_billing/domain/services/i_meter.py
Python
async def record(self, record: UsageRecord) -> None: ...

aggregate async

Python
aggregate(tenant_id: str, metric: str, since_s: float = 0.0) -> float
Source code in apogee_ai_billing/domain/services/i_meter.py
Python
async def aggregate(
    self, tenant_id: str, metric: str, since_s: float = 0.0
) -> float: ...

list_records async

Python
list_records(tenant_id: str, since_s: float = 0.0) -> Iterable[UsageRecord]
Source code in apogee_ai_billing/domain/services/i_meter.py
Python
async def list_records(
    self, tenant_id: str, since_s: float = 0.0
) -> Iterable[UsageRecord]: ...

IPlanRepository

Bases: Protocol

register

Python
register(plan: Plan) -> None
Source code in apogee_ai_billing/domain/services/i_plan_repository.py
Python
def register(self, plan: Plan) -> None: ...

get

Python
get(code: str) -> Plan
Source code in apogee_ai_billing/domain/services/i_plan_repository.py
Python
def get(self, code: str) -> Plan: ...

list

Python
list() -> Iterable[Plan]
Source code in apogee_ai_billing/domain/services/i_plan_repository.py
Python
def list(self) -> Iterable[Plan]: ...

ISubscriptionRepository

Bases: Protocol

upsert async

Python
upsert(subscription: Subscription) -> None
Source code in apogee_ai_billing/domain/services/i_subscription_repository.py
Python
async def upsert(self, subscription: Subscription) -> None: ...

get async

Python
get(tenant_id: str) -> Subscription
Source code in apogee_ai_billing/domain/services/i_subscription_repository.py
Python
async def get(self, tenant_id: str) -> Subscription: ...

cancel async

Python
cancel(tenant_id: str) -> None
Source code in apogee_ai_billing/domain/services/i_subscription_repository.py
Python
async def cancel(self, tenant_id: str) -> None: ...

Infrastructure

CompositeMeter

Python
CompositeMeter(meters: dict[str, object])

Aggregates several meters under one façade. Routes by metric name.

Source code in apogee_ai_billing/infrastructure/meters/composite_meter.py
Python
def __init__(self, meters: dict[str, object]) -> None:
    if not meters:
        raise ValueError("at least one meter is required")
    self._meters = meters

name class-attribute instance-attribute

Python
name = 'composite'

record async

Python
record(record: UsageRecord) -> None
Source code in apogee_ai_billing/infrastructure/meters/composite_meter.py
Python
async def record(self, record: UsageRecord) -> None:
    meter = self._meters.get(record.metric) or next(iter(self._meters.values()))
    await meter.record(record)  # type: ignore[union-attr]

aggregate async

Python
aggregate(tenant_id: str, metric: str, since_s: float = 0.0) -> float
Source code in apogee_ai_billing/infrastructure/meters/composite_meter.py
Python
async def aggregate(
    self, tenant_id: str, metric: str, since_s: float = 0.0
) -> float:
    meter = self._meters.get(metric)
    if meter is None:
        return 0.0
    return await meter.aggregate(tenant_id, metric, since_s=since_s)  # type: ignore[union-attr]

list_records async

Python
list_records(tenant_id: str, since_s: float = 0.0) -> Iterable[UsageRecord]
Source code in apogee_ai_billing/infrastructure/meters/composite_meter.py
Python
async def list_records(
    self, tenant_id: str, since_s: float = 0.0
) -> Iterable[UsageRecord]:
    out: list[UsageRecord] = []
    for meter in self._meters.values():
        out.extend(await meter.list_records(tenant_id, since_s=since_s))  # type: ignore[union-attr]
    return out

FlatPlan

Python
FlatPlan(*, code: str, name: str, monthly_usd: float, included_tokens: int = 0, overage_per_1k_usd: float = 0.0) -> Plan
Source code in apogee_ai_billing/infrastructure/plans/factories.py
Python
def FlatPlan(
    *,
    code: str,
    name: str,
    monthly_usd: float,
    included_tokens: int = 0,
    overage_per_1k_usd: float = 0.0,
) -> Plan:
    return Plan(
        code=code,
        name=name,
        kind=PlanKind.FLAT,
        period=BillingPeriod.MONTHLY,
        base_fee_usd=monthly_usd,
        included_units=included_tokens,
        overage_price_per_1k_usd=overage_per_1k_usd,
    )

InMemoryInvoiceRepository

Python
InMemoryInvoiceRepository()
Source code in apogee_ai_billing/infrastructure/registries/in_memory_invoice_repository.py
Python
def __init__(self) -> None:
    self._invoices: dict[str, Invoice] = {}

name class-attribute instance-attribute

Python
name = 'in_memory_invoice'

save async

Python
save(invoice: Invoice) -> None
Source code in apogee_ai_billing/infrastructure/registries/in_memory_invoice_repository.py
Python
async def save(self, invoice: Invoice) -> None:
    self._invoices[invoice.id] = invoice

get async

Python
get(invoice_id: str) -> Invoice
Source code in apogee_ai_billing/infrastructure/registries/in_memory_invoice_repository.py
Python
async def get(self, invoice_id: str) -> Invoice:
    if invoice_id not in self._invoices:
        raise InvoiceNotFoundException(f"invoice not found: {invoice_id!r}")
    return self._invoices[invoice_id]

list_by_tenant async

Python
list_by_tenant(tenant_id: str) -> Iterable[Invoice]
Source code in apogee_ai_billing/infrastructure/registries/in_memory_invoice_repository.py
Python
async def list_by_tenant(self, tenant_id: str) -> Iterable[Invoice]:
    return [inv for inv in self._invoices.values() if inv.tenant_id == tenant_id]

count async

Python
count() -> int
Source code in apogee_ai_billing/infrastructure/registries/in_memory_invoice_repository.py
Python
async def count(self) -> int:
    return len(self._invoices)

InMemoryMeter

Python
InMemoryMeter()

In-memory meter. Records flat list, aggregates on demand.

Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
def __init__(self) -> None:
    self._records: list[UsageRecord] = []

name class-attribute instance-attribute

Python
name = 'in_memory'

record async

Python
record(record: UsageRecord) -> None
Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
async def record(self, record: UsageRecord) -> None:
    self._records.append(record)

aggregate async

Python
aggregate(tenant_id: str, metric: str, since_s: float = 0.0) -> float
Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
async def aggregate(
    self, tenant_id: str, metric: str, since_s: float = 0.0
) -> float:
    return sum(
        r.quantity
        for r in self._records
        if r.tenant_id == tenant_id
        and r.metric == metric
        and r.timestamp_s >= since_s
    )

list_records async

Python
list_records(tenant_id: str, since_s: float = 0.0) -> Iterable[UsageRecord]
Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
async def list_records(
    self, tenant_id: str, since_s: float = 0.0
) -> Iterable[UsageRecord]:
    return [
        r for r in self._records
        if r.tenant_id == tenant_id and r.timestamp_s >= since_s
    ]

InMemoryPlanRepository

Python
InMemoryPlanRepository(plans: Iterable[Plan] = ())
Source code in apogee_ai_billing/infrastructure/registries/in_memory_plan_repository.py
Python
def __init__(self, plans: Iterable[Plan] = ()) -> None:
    self._plans: dict[str, Plan] = {p.code: p for p in plans}

name class-attribute instance-attribute

Python
name = 'in_memory_plan'

register

Python
register(plan: Plan) -> None
Source code in apogee_ai_billing/infrastructure/registries/in_memory_plan_repository.py
Python
def register(self, plan: Plan) -> None:
    self._plans[plan.code] = plan

get

Python
get(code: str) -> Plan
Source code in apogee_ai_billing/infrastructure/registries/in_memory_plan_repository.py
Python
def get(self, code: str) -> Plan:
    if code not in self._plans:
        raise PlanNotFoundException(code)
    return self._plans[code]

list

Python
list() -> list[Plan]
Source code in apogee_ai_billing/infrastructure/registries/in_memory_plan_repository.py
Python
def list(self) -> list[Plan]:
    return list(self._plans.values())

InMemorySubscriptionRepository

Python
InMemorySubscriptionRepository()
Source code in apogee_ai_billing/infrastructure/registries/in_memory_subscription_repository.py
Python
def __init__(self) -> None:
    self._subs: dict[str, Subscription] = {}

name class-attribute instance-attribute

Python
name = 'in_memory_subscription'

upsert async

Python
upsert(subscription: Subscription) -> None
Source code in apogee_ai_billing/infrastructure/registries/in_memory_subscription_repository.py
Python
async def upsert(self, subscription: Subscription) -> None:
    self._subs[subscription.tenant_id] = subscription

get async

Python
get(tenant_id: str) -> Subscription
Source code in apogee_ai_billing/infrastructure/registries/in_memory_subscription_repository.py
Python
async def get(self, tenant_id: str) -> Subscription:
    if tenant_id not in self._subs:
        raise SubscriptionNotFoundException(tenant_id)
    return self._subs[tenant_id]

cancel async

Python
cancel(tenant_id: str) -> None
Source code in apogee_ai_billing/infrastructure/registries/in_memory_subscription_repository.py
Python
async def cancel(self, tenant_id: str) -> None:
    if tenant_id not in self._subs:
        raise SubscriptionNotFoundException(tenant_id)
    self._subs[tenant_id] = replace(self._subs[tenant_id], active=False)

InvoiceBuilder

Python
InvoiceBuilder()

Builds an Invoice from a Plan + usage volume for the period.

Source code in apogee_ai_billing/infrastructure/invoices/invoice_builder.py
Python
def __init__(self) -> None:
    pass

build

Python
build(*, tenant_id: str, plan: Plan, usage_units: float, period_start_s: float, period_end_s: float | None = None) -> Invoice
Source code in apogee_ai_billing/infrastructure/invoices/invoice_builder.py
Python
def build(
    self,
    *,
    tenant_id: str,
    plan: Plan,
    usage_units: float,
    period_start_s: float,
    period_end_s: float | None = None,
) -> Invoice:
    if usage_units < 0:
        raise ValueError("usage_units must be non-negative")
    end = period_end_s if period_end_s is not None else time.time()
    lines: list[InvoiceLine] = []
    if plan.base_fee_usd > 0:
        lines.append(_line(
            description=f"{plan.name} base fee ({plan.period.value})",
            quantity=1,
            unit_price_usd=plan.base_fee_usd,
        ))
    if plan.kind is PlanKind.FLAT:
        overage = max(0.0, usage_units - plan.included_units)
        if overage > 0 and plan.overage_price_per_1k_usd > 0:
            unit = plan.overage_price_per_1k_usd / 1000.0
            lines.append(_line(
                description=(
                    f"Overage ({int(overage):,} units beyond {plan.included_units:,})"
                ),
                quantity=overage,
                unit_price_usd=unit,
            ))
    elif plan.kind is PlanKind.TIERED:
        lines.extend(_tier_lines(plan.tiers, usage_units))
    elif plan.kind is PlanKind.USAGE_BASED:
        unit = plan.overage_price_per_1k_usd / 1000.0
        if usage_units > 0 and unit > 0:
            lines.append(_line(
                description=f"Usage ({int(usage_units):,} units)",
                quantity=usage_units,
                unit_price_usd=unit,
            ))
    return Invoice(
        tenant_id=tenant_id,
        plan_code=plan.code,
        period_start_s=period_start_s,
        period_end_s=end,
        lines=tuple(lines),
        status=InvoiceStatus.OPEN,
        metadata={"usage_units": str(int(usage_units))},
    )

JsonInvoiceFormatter

name class-attribute instance-attribute

Python
name = 'json'

format

Python
format(invoice: Invoice) -> str
Source code in apogee_ai_billing/infrastructure/invoices/formatters.py
Python
def format(self, invoice: Invoice) -> str:
    payload = {
        "id": invoice.id,
        "tenant_id": invoice.tenant_id,
        "plan_code": invoice.plan_code,
        "status": invoice.status.value,
        "period_start_s": invoice.period_start_s,
        "period_end_s": invoice.period_end_s,
        "issued_at_s": invoice.issued_at_s,
        "total_usd": round(invoice.total_usd, 6),
        "lines": [
            {
                "description": line.description,
                "quantity": line.quantity,
                "unit_price_usd": round(line.unit_price_usd, 8),
                "amount_usd": round(line.amount_usd, 6),
            }
            for line in invoice.lines
        ],
        "metadata": dict(invoice.metadata),
    }
    return json.dumps(payload, indent=2, default=str)

MarkdownInvoiceFormatter

name class-attribute instance-attribute

Python
name = 'markdown'

format

Python
format(invoice: Invoice) -> str
Source code in apogee_ai_billing/infrastructure/invoices/formatters.py
Python
def format(self, invoice: Invoice) -> str:
    lines = [
        f"# Invoice `{invoice.id}`",
        "",
        f"- **Tenant**: `{invoice.tenant_id}`",
        f"- **Plan**: `{invoice.plan_code}`",
        f"- **Status**: {invoice.status.value}",
        f"- **Total**: ${invoice.total_usd:.6f}",
        "",
        "| Description | Quantity | Unit price | Amount |",
        "|---|---:|---:|---:|",
    ]
    for line in invoice.lines:
        lines.append(
            f"| {line.description} | {line.quantity:,.2f} | "
            f"${line.unit_price_usd:.8f} | ${line.amount_usd:.6f} |"
        )
    return "\n".join(lines)

RequestMeter

Python
RequestMeter()

Bases: InMemoryMeter

Convenience meter typed for the requests metric.

Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
def __init__(self) -> None:
    self._records: list[UsageRecord] = []

name class-attribute instance-attribute

Python
name = 'requests'

record_request async

Python
record_request(tenant_id: str, count: int = 1, **metadata: str) -> UsageRecord
Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
async def record_request(
    self, tenant_id: str, count: int = 1, **metadata: str
) -> UsageRecord:
    record = UsageRecord(
        tenant_id=tenant_id,
        metric="requests",
        quantity=float(count),
        metadata={k: str(v) for k, v in metadata.items()},
    )
    await self.record(record)
    return record

TieredPlan

Python
TieredPlan(*, code: str, name: str, monthly_usd: float = 0.0, tiers: tuple[PriceTier, ...]) -> Plan
Source code in apogee_ai_billing/infrastructure/plans/factories.py
Python
def TieredPlan(
    *,
    code: str,
    name: str,
    monthly_usd: float = 0.0,
    tiers: tuple[PriceTier, ...],
) -> Plan:
    return Plan(
        code=code,
        name=name,
        kind=PlanKind.TIERED,
        period=BillingPeriod.MONTHLY,
        base_fee_usd=monthly_usd,
        tiers=tiers,
    )

TokenMeter

Python
TokenMeter()

Bases: InMemoryMeter

Convenience meter typed for the tokens metric.

Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
def __init__(self) -> None:
    self._records: list[UsageRecord] = []

name class-attribute instance-attribute

Python
name = 'tokens'

record_tokens async

Python
record_tokens(tenant_id: str, tokens: int, **metadata: str) -> UsageRecord
Source code in apogee_ai_billing/infrastructure/meters/in_memory_meter.py
Python
async def record_tokens(
    self, tenant_id: str, tokens: int, **metadata: str
) -> UsageRecord:
    record = UsageRecord(
        tenant_id=tenant_id,
        metric="tokens",
        quantity=float(tokens),
        metadata={k: str(v) for k, v in metadata.items()},
    )
    await self.record(record)
    return record

UsagePlan

Python
UsagePlan(*, code: str, name: str, overage_per_1k_usd: float) -> Plan
Source code in apogee_ai_billing/infrastructure/plans/factories.py
Python
def UsagePlan(
    *,
    code: str,
    name: str,
    overage_per_1k_usd: float,
) -> Plan:
    return Plan(
        code=code,
        name=name,
        kind=PlanKind.USAGE_BASED,
        period=BillingPeriod.MONTHLY,
        base_fee_usd=0.0,
        included_units=0,
        overage_price_per_1k_usd=overage_per_1k_usd,
    )