feat(stats): per-user API quota attribution + admin usage page
Track who burned how much YouTube API quota. A QuotaEvent audit log (migration 0009) records every spend with the triggering user (NULL = background/system) and an action label, set via a request/job-scoped contextvar (quota.attribute) so no call signatures change. User-initiated work (sync subscriptions, unsubscribe, opt-in recent backfill, manual enrich) attributes to the user; scheduler work to System, split by action. - backend: QuotaEvent model + migration 0009; quota.attribute() contextvar; record_usage logs events; entry points wrapped (routes/sync, routes/channels, scheduler); GET /api/quota/my-usage + GET /api/quota/admin - frontend: admin-only Stats page (header nav, page=stats) with daily bars + per-user breakdown by action and range picker; 'Your API usage' in Settings -> Sync for every user Verified: attribution + endpoints compute correctly; events are per-user vs System.
This commit is contained in:
parent
0dad71cf6e
commit
d610647d13
15 changed files with 451 additions and 26 deletions
|
|
@ -26,7 +26,7 @@ from starlette.middleware.sessions import SessionMiddleware
|
|||
|
||||
from app import auth
|
||||
from app.config import settings
|
||||
from app.routes import admin, channels, feed, health, me, sync, tags
|
||||
from app.routes import admin, channels, feed, health, me, quota, sync, tags
|
||||
from app.scheduler import shutdown_scheduler, start_scheduler
|
||||
|
||||
|
||||
|
|
@ -67,6 +67,7 @@ app.include_router(feed.router)
|
|||
app.include_router(me.router)
|
||||
app.include_router(channels.router)
|
||||
app.include_router(admin.router)
|
||||
app.include_router(quota.router)
|
||||
|
||||
# The built SPA (populated by the Docker frontend build stage).
|
||||
STATIC_DIR = Path(__file__).parent / "static_spa"
|
||||
|
|
|
|||
|
|
@ -265,6 +265,25 @@ class ApiQuotaUsage(Base):
|
|||
units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
|
||||
|
||||
|
||||
class QuotaEvent(Base):
|
||||
"""Per-spend audit log for YouTube API quota attribution. `user_id` is the end-user
|
||||
whose action caused the spend; NULL means background/system work (scheduler backfill,
|
||||
enrichment, resync). The aggregate budget still lives in ApiQuotaUsage; this is detail."""
|
||||
|
||||
__tablename__ = "quota_events"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
# SET NULL on user delete: keep the audit trail (it just becomes system-attributed).
|
||||
user_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="SET NULL"), index=True
|
||||
)
|
||||
action: Mapped[str] = mapped_column(String(40), index=True)
|
||||
units: Mapped[int] = mapped_column(Integer)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), index=True
|
||||
)
|
||||
|
||||
|
||||
class AppState(Base):
|
||||
"""Single-row global app state (admin-controlled)."""
|
||||
|
||||
|
|
|
|||
|
|
@ -4,22 +4,52 @@ The whole app shares one daily budget (the API resets at midnight US Pacific tim
|
|||
Steady-state work (RSS detection is free; enrichment of new videos is cheap) should
|
||||
always be allowed; expensive backfill must yield to the remaining budget.
|
||||
"""
|
||||
from datetime import datetime
|
||||
import contextlib
|
||||
import contextvars
|
||||
from datetime import datetime, timezone
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import settings
|
||||
from app.models import ApiQuotaUsage
|
||||
from app.models import ApiQuotaUsage, QuotaEvent
|
||||
|
||||
_PACIFIC = ZoneInfo("America/Los_Angeles")
|
||||
|
||||
# Request/job-scoped attribution for quota spend: who triggered it and what kind of work.
|
||||
# Set at entry points (route handlers, scheduler jobs) via attribute(); read by
|
||||
# record_usage. Default = background/system, generic action.
|
||||
_actor_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
|
||||
"quota_actor_id", default=None
|
||||
)
|
||||
_action: contextvars.ContextVar[str] = contextvars.ContextVar(
|
||||
"quota_action", default="api"
|
||||
)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def attribute(actor_id: int | None, action: str):
|
||||
"""Attribute any quota spent in this block to `actor_id` (None = system) under `action`."""
|
||||
t1 = _actor_id.set(actor_id)
|
||||
t2 = _action.set(action)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_actor_id.reset(t1)
|
||||
_action.reset(t2)
|
||||
|
||||
|
||||
def pacific_today():
|
||||
return datetime.now(_PACIFIC).date()
|
||||
|
||||
|
||||
def pacific_day_start_utc() -> datetime:
|
||||
"""UTC instant of the current Pacific day's midnight — aligns 'today' with the budget reset."""
|
||||
start = datetime.now(_PACIFIC).replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
return start.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def units_used_today(db: Session) -> int:
|
||||
row = db.get(ApiQuotaUsage, pacific_today())
|
||||
return row.units_used if row else 0
|
||||
|
|
@ -34,7 +64,7 @@ def can_spend(db: Session, units: int) -> bool:
|
|||
|
||||
|
||||
def record_usage(db: Session, units: int) -> None:
|
||||
"""Atomically add `units` to today's counter (upsert)."""
|
||||
"""Atomically add `units` to today's counter (upsert) and log an attribution event."""
|
||||
if units <= 0:
|
||||
return
|
||||
day = pacific_today()
|
||||
|
|
@ -47,4 +77,6 @@ def record_usage(db: Session, units: int) -> None:
|
|||
)
|
||||
)
|
||||
db.execute(stmt)
|
||||
# Audit detail: who/what spent it (per-user attribution; NULL actor = system).
|
||||
db.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units))
|
||||
db.commit()
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException
|
|||
from sqlalchemy import and_, func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import quota
|
||||
from app.auth import current_user, has_write_scope
|
||||
from app.db import get_db
|
||||
from app.models import Channel, ChannelTag, Subscription, Tag, User, Video
|
||||
|
|
@ -121,7 +122,8 @@ def update_channel(
|
|||
# fetched its recent uploads yet, do that one channel now (cheap) instead of waiting for
|
||||
# the scheduler. Deep paging still follows on the scheduler's next run (recent-then-deep).
|
||||
if deep_turned_on and channel is not None and channel.recent_synced_at is None:
|
||||
run_recent_backfill(db, [channel], max_channels=1)
|
||||
with quota.attribute(user.id, "backfill_recent"):
|
||||
run_recent_backfill(db, [channel], max_channels=1)
|
||||
|
||||
return {
|
||||
"id": channel_id,
|
||||
|
|
@ -152,7 +154,7 @@ def unsubscribe(
|
|||
detail="No YouTube subscription id on record — sync subscriptions first.",
|
||||
)
|
||||
try:
|
||||
with YouTubeClient(db, user) as yt:
|
||||
with quota.attribute(user.id, "unsubscribe"), YouTubeClient(db, user) as yt:
|
||||
yt.delete_subscription(sub.yt_subscription_id)
|
||||
except YouTubeError as exc:
|
||||
raise HTTPException(status_code=502, detail=f"YouTube unsubscribe failed: {exc}")
|
||||
|
|
|
|||
106
backend/app/routes/quota.py
Normal file
106
backend/app/routes/quota.py
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
"""Quota usage attribution: a user's own consumption + an admin breakdown across users.
|
||||
|
||||
The aggregate daily budget lives in ApiQuotaUsage; here we read the per-spend audit log
|
||||
(QuotaEvent) to attribute units to the user who triggered them (NULL actor = system).
|
||||
"""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import quota
|
||||
from app.auth import current_user
|
||||
from app.db import get_db
|
||||
from app.models import QuotaEvent, User
|
||||
from app.routes.admin import admin_user
|
||||
|
||||
router = APIRouter(prefix="/api/quota", tags=["quota"])
|
||||
|
||||
|
||||
def _sum(db: Session, *conds) -> int:
|
||||
return int(
|
||||
db.scalar(
|
||||
select(func.coalesce(func.sum(QuotaEvent.units), 0)).where(*conds)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
|
||||
|
||||
@router.get("/my-usage")
|
||||
def my_usage(
|
||||
user: User = Depends(current_user), db: Session = Depends(get_db)
|
||||
) -> dict:
|
||||
now = datetime.now(timezone.utc)
|
||||
mine = QuotaEvent.user_id == user.id
|
||||
by_action = {
|
||||
action: int(units)
|
||||
for action, units in db.execute(
|
||||
select(QuotaEvent.action, func.sum(QuotaEvent.units))
|
||||
.where(mine, QuotaEvent.created_at >= now - timedelta(days=30))
|
||||
.group_by(QuotaEvent.action)
|
||||
).all()
|
||||
}
|
||||
return {
|
||||
"today": _sum(db, mine, QuotaEvent.created_at >= quota.pacific_day_start_utc()),
|
||||
"last_7d": _sum(db, mine, QuotaEvent.created_at >= now - timedelta(days=7)),
|
||||
"last_30d": _sum(db, mine, QuotaEvent.created_at >= now - timedelta(days=30)),
|
||||
"all_time": _sum(db, mine),
|
||||
"by_action": by_action,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/admin")
|
||||
def admin_usage(
|
||||
days: int = 30,
|
||||
_: User = Depends(admin_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
days = max(1, min(days, 365))
|
||||
since = datetime.now(timezone.utc) - timedelta(days=days)
|
||||
|
||||
totals = db.execute(
|
||||
select(QuotaEvent.user_id, func.sum(QuotaEvent.units))
|
||||
.where(QuotaEvent.created_at >= since)
|
||||
.group_by(QuotaEvent.user_id)
|
||||
).all()
|
||||
|
||||
per_action: dict[int | None, dict[str, int]] = {}
|
||||
for uid, action, units in db.execute(
|
||||
select(QuotaEvent.user_id, QuotaEvent.action, func.sum(QuotaEvent.units))
|
||||
.where(QuotaEvent.created_at >= since)
|
||||
.group_by(QuotaEvent.user_id, QuotaEvent.action)
|
||||
).all():
|
||||
per_action.setdefault(uid, {})[action] = int(units)
|
||||
|
||||
uids = [uid for uid, _ in totals if uid is not None]
|
||||
emails = (
|
||||
{u.id: u.email for u in db.execute(select(User).where(User.id.in_(uids))).scalars()}
|
||||
if uids
|
||||
else {}
|
||||
)
|
||||
|
||||
rows = [
|
||||
{
|
||||
"user_id": uid,
|
||||
"email": emails.get(uid, "(deleted user)") if uid is not None else "System",
|
||||
"total": int(total),
|
||||
"by_action": per_action.get(uid, {}),
|
||||
}
|
||||
for uid, total in totals
|
||||
]
|
||||
rows.sort(key=lambda r: r["total"], reverse=True)
|
||||
|
||||
# Instance-wide daily series in Pacific days (matches the budget reset).
|
||||
day_expr = func.date(func.timezone("America/Los_Angeles", QuotaEvent.created_at))
|
||||
daily = [
|
||||
{"day": str(day), "total": int(units)}
|
||||
for day, units in db.execute(
|
||||
select(day_expr, func.sum(QuotaEvent.units))
|
||||
.where(QuotaEvent.created_at >= since)
|
||||
.group_by(day_expr)
|
||||
.order_by(day_expr)
|
||||
).all()
|
||||
]
|
||||
|
||||
return {"range_days": days, "rows": rows, "daily": daily}
|
||||
|
|
@ -34,7 +34,8 @@ def sync_subscriptions(
|
|||
user: User = Depends(current_user), db: Session = Depends(get_db)
|
||||
) -> dict:
|
||||
before = quota.units_used_today(db)
|
||||
result = import_subscriptions(db, user)
|
||||
with quota.attribute(user.id, "sync_subscriptions"):
|
||||
result = import_subscriptions(db, user)
|
||||
result["quota_used_estimate"] = quota.units_used_today(db) - before
|
||||
result["quota_remaining_today"] = quota.remaining_today(db)
|
||||
return result
|
||||
|
|
@ -55,7 +56,8 @@ def sync_backfill(
|
|||
max_channels: int = 25,
|
||||
) -> dict:
|
||||
before = quota.units_used_today(db)
|
||||
result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels)
|
||||
with quota.attribute(user.id, "backfill_recent"):
|
||||
result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels)
|
||||
result["quota_used_estimate"] = quota.units_used_today(db) - before
|
||||
return result
|
||||
|
||||
|
|
@ -65,7 +67,8 @@ def sync_enrich(
|
|||
user: User = Depends(current_user), db: Session = Depends(get_db)
|
||||
) -> dict:
|
||||
before = quota.units_used_today(db)
|
||||
enriched = run_enrich(db)
|
||||
with quota.attribute(user.id, "enrich"):
|
||||
enriched = run_enrich(db)
|
||||
return {
|
||||
"enriched": enriched,
|
||||
"quota_used_estimate": quota.units_used_today(db) - before,
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import logging
|
|||
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
|
||||
from app import quota
|
||||
from app.config import settings
|
||||
from app.db import SessionLocal
|
||||
from app.state import is_sync_paused
|
||||
|
|
@ -42,14 +43,21 @@ def _rss_job() -> None:
|
|||
|
||||
|
||||
def _enrich_job() -> None:
|
||||
_job("enrich", lambda db: run_enrich(db))
|
||||
def work(db):
|
||||
with quota.attribute(None, "enrich"):
|
||||
return run_enrich(db)
|
||||
|
||||
_job("enrich", work)
|
||||
|
||||
|
||||
def _backfill_job() -> None:
|
||||
# Recent-first for not-yet-synced channels, then deep backfill for the rest.
|
||||
# Recent-first for not-yet-synced channels, then deep backfill for the rest. All
|
||||
# background spend is attributed to the system (no actor), split by action.
|
||||
def work(db):
|
||||
recent = run_recent_backfill(db, max_channels=25)
|
||||
deep = run_deep_backfill(db, max_channels=10)
|
||||
with quota.attribute(None, "backfill_recent"):
|
||||
recent = run_recent_backfill(db, max_channels=25)
|
||||
with quota.attribute(None, "backfill_deep"):
|
||||
deep = run_deep_backfill(db, max_channels=10)
|
||||
return {"recent": recent, "deep": deep}
|
||||
|
||||
_job("backfill", work)
|
||||
|
|
@ -64,7 +72,11 @@ def _shorts_job() -> None:
|
|||
|
||||
|
||||
def _subscriptions_job() -> None:
|
||||
_job("subscriptions", run_subscription_resync)
|
||||
def work(db):
|
||||
with quota.attribute(None, "subscription_resync"):
|
||||
return run_subscription_resync(db)
|
||||
|
||||
_job("subscriptions", work)
|
||||
|
||||
|
||||
def start_scheduler() -> None:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue