Misurazione dell’utilizzo, quote e hook di fatturazione

from dataclasses import dataclass

@dataclass(frozen=True)
class Quota:
    metric: str
    limit: int | None   # None = unlimited
    period: str          # 'day' or 'month'

PLANS = {
    "free": [Quota("api_calls", 1_000, "month"), Quota("ai_tokens", 50_000, "month")],
    "pro": [Quota("api_calls", 100_000, "month"), Quota("ai_tokens", 5_000_000, "month")],
    "enterprise": [Quota("api_calls", None, "month"), Quota("ai_tokens", None, "month")],
}

def limit_for(plan: str, metric: str) -> int | None:
    for q in PLANS[plan]:
        if q.metric == metric:
            return q.limit
    return 0  # metric not allowed on this plan

print(limit_for("free", "api_calls"))   # 1000
print(limit_for("enterprise", "ai_tokens"))  # None (unlimited)
print(limit_for("free", "unknown"))     # 0

from collections import defaultdict
from datetime import datetime, timezone

usage = defaultdict(int)

def period_key(period: str, now: datetime) -> str:
    if period == "month":
        return now.strftime("%Y-%m")
    return now.strftime("%Y-%m-%d")

def incr(tenant: str, metric: str, amount: int, period: str, now: datetime) -> int:
    key = (tenant, metric, period_key(period, now))
    usage[key] += amount
    return usage[key]

now = datetime(2026, 6, 10, tzinfo=timezone.utc)
print(incr("acme", "api_calls", 1, "month", now))  # 1
print(incr("acme", "api_calls", 1, "month", now))  # 2
print(incr("acme", "ai_tokens", 1200, "month", now))  # 1200

def check_quota(current: int, amount: int, limit: int | None) -> bool:
    """Return True if `amount` more units are allowed."""
    if limit is None:
        return True  # unlimited
    return current + amount <= limit

print(check_quota(995, 1, 1000))    # True  (996 <= 1000)
print(check_quota(1000, 1, 1000))   # False (1001 > 1000)
print(check_quota(9_000_000, 1, None))  # True (unlimited)

from fastapi import Depends, HTTPException, Request

async def enforce_quota(metric: str, cost: int = 1):
    async def _dep(request: Request):
        tenant = request.state.tenant_id
        plan = await get_plan(tenant)
        limit = limit_for(plan, metric)
        current = await store.get(tenant, metric, "month")
        if not check_quota(current, cost, limit):
            raise HTTPException(
                status_code=429,
                detail=f"Quota exceeded for {metric}",
                headers={"X-Quota-Limit": str(limit)},
            )
        await store.incr(tenant, metric, cost, "month")
        return current + cost
    return _dep

@app.post("/v1/complete")
async def complete(used=Depends(enforce_quota("ai_tokens", cost=500))):
    return {"ok": True, "tokens_used_this_month": used}

from contextlib import contextmanager

@contextmanager
def meter(record, tenant, metric):
    actual = {"amount": 0}
    try:
        yield actual            # caller sets actual['amount']
    finally:
        if actual["amount"]:
            record(tenant, metric, actual["amount"])

log = []
with meter(lambda t, m, a: log.append((t, m, a)), "acme", "ai_tokens") as use:
    # ... do real work, then we learn the true cost ...
    use["amount"] = 1342

print(log)  # [('acme', 'ai_tokens', 1342)]

def classify(current: int, limit: int | None, overage_allowed: bool) -> str:
    if limit is None:
        return "ok"
    if current < 0.8 * limit:
        return "ok"
    if current < limit:
        return "warn"
    return "overage" if overage_allowed else "blocked"

print(classify(700, 1000, False))   # ok
print(classify(850, 1000, False))   # warn
print(classify(1200, 1000, True))   # overage
print(classify(1200, 1000, False))  # blocked

import uuid, json
from datetime import datetime, timezone

def billing_event(tenant: str, metric: str, qty: int, ts: datetime, request_id: str) -> dict:
    return {
        "id": str(uuid.uuid4()),
        "idempotency_key": f"{tenant}:{metric}:{request_id}",
        "tenant_id": tenant,
        "metric": metric,
        "quantity": qty,
        "occurred_at": ts.isoformat(),
    }

evt = billing_event("acme", "ai_tokens", 1342,
                    datetime(2026, 6, 10, tzinfo=timezone.utc), "req_9f3")
print(json.dumps(evt, indent=2))

from fastapi import BackgroundTasks

@app.post("/v1/complete")
async def complete(bg: BackgroundTasks, request: Request):
    tenant = request.state.tenant_id
    result, tokens = await run_completion(request)
    await store.incr(tenant, "ai_tokens", tokens, "month")
    evt = billing_event(tenant, "ai_tokens", tokens,
                        datetime.now(timezone.utc), request.state.request_id)
    bg.add_task(emit_to_billing, evt)  # fire-and-forget, off the hot path
    return result

from collections import defaultdict

def reconcile(events: list[dict]) -> dict:
    """Rebuild authoritative totals from the immutable event log."""
    totals = defaultdict(int)
    seen = set()
    for e in events:
        if e["idempotency_key"] in seen:
            continue  # dedupe replays
        seen.add(e["idempotency_key"])
        period = e["occurred_at"][:7]  # YYYY-MM
        totals[(e["tenant_id"], e["metric"], period)] += e["quantity"]
    return dict(totals)

events = [
    {"idempotency_key": "acme:t:1", "tenant_id": "acme", "metric": "t", "quantity": 100, "occurred_at": "2026-06-01T10:00:00"},
    {"idempotency_key": "acme:t:1", "tenant_id": "acme", "metric": "t", "quantity": 100, "occurred_at": "2026-06-01T10:00:00"},
    {"idempotency_key": "acme:t:2", "tenant_id": "acme", "metric": "t", "quantity": 50, "occurred_at": "2026-06-02T09:00:00"},
]
print(reconcile(events))  # {('acme','t','2026-06'): 150}