T11.3: usage aggregation endpoint with filters (CR-019)

GET /api/usage/summary: active-user counts by day/week/month, per-user
last-active, per-tool breakdown. Filters (from/to/project_id/username/
tool) combine. Gated by require_user_manager - same boundary as the User
Directory. A project_super_user is scoped to events tied to projects
they manage plus their own activity (managed_project_ids), never another
user's suite-wide activity outside that; an app admin sees everything.

_usage_query() factored out so T11.4's export can never disagree with
what this endpoint counted - same filtered row set, not two derivations.

Verified: admin sees all seeded events; a project_super_user scoped to
one of two projects correctly sees only that project's events plus their
own account-wide activity, and specifically does NOT see the admin's
other-project or no-project activity; date/tool/project filters each
narrow results correctly and combine.
This commit is contained in:
2026-09-23 12:19:14 -07:00
parent 6cde6e3f60
commit 8e863ae7d0

View File

@@ -13,6 +13,7 @@ import logging
import os
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
from urllib.parse import urlparse
@@ -22,7 +23,7 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import select, delete, func
from sqlalchemy import select, delete, func, or_
from sqlalchemy.orm import Session
from starlette.middleware.sessions import SessionMiddleware
@@ -3413,6 +3414,101 @@ def usage_ping(body: UsageIn, user: models.User = Depends(auth.get_current_user)
return {"ok": True}
def _parse_date_q(v: str, end: bool = False) -> Optional[datetime]:
"""Accepts a plain YYYY-MM-DD (what a <input type=date> sends) or a full
ISO datetime. A date-only `to` means "through the end of that day", not
midnight at its start - otherwise a range of "today" would match nothing
from today at all."""
if not v:
return None
try:
d = datetime.fromisoformat(v)
except ValueError:
return None
if d.tzinfo is None:
d = d.replace(tzinfo=timezone.utc)
if end and len(v) <= 10: # date-only
d = d + timedelta(days=1) - timedelta(microseconds=1)
return d
def _usage_query(db: Session, caller: "models.User", date_from, date_to, project_id, username, tool):
"""Shared by the summary and export endpoints so the two can never
disagree about which rows a filter set matches - the export is a raw
dump of exactly what the summary counted, not a separately-derived view."""
stmt = select(models.UsageEvent)
managed = managed_project_ids(db, caller)
if managed is not None:
# A project_super_user (never an app admin, who gets managed=None) is
# scoped to events tied to a project they administer, plus their OWN
# suite-wide activity (admin console opens etc. carry no project_id) -
# never another user's activity outside what they manage.
if managed:
stmt = stmt.where(or_(
models.UsageEvent.project_id.in_(managed),
models.UsageEvent.username == caller.username,
))
else:
stmt = stmt.where(models.UsageEvent.username == caller.username)
df = _parse_date_q(date_from)
dt = _parse_date_q(date_to, end=True)
if df:
stmt = stmt.where(models.UsageEvent.at >= df)
if dt:
stmt = stmt.where(models.UsageEvent.at <= dt)
if project_id:
stmt = stmt.where(models.UsageEvent.project_id == project_id)
if username:
stmt = stmt.where(models.UsageEvent.username == username)
if tool:
stmt = stmt.where(models.UsageEvent.tool == tool)
return stmt.order_by(models.UsageEvent.at)
@app.get("/api/usage/summary")
def usage_summary(
date_from: Optional[str] = Query(None, alias="from"),
date_to: Optional[str] = Query(None, alias="to"),
project_id: Optional[str] = Query(None),
username: Optional[str] = Query(None),
tool: Optional[str] = Query(None),
caller: models.User = Depends(require_user_manager),
db: Session = Depends(get_db),
):
"""CR-019. Same gate as the User Directory (require_user_manager): an app
admin or a project_super_user with at least one managed project. Filters
combine. Aggregated in Python over the filtered row set rather than a SQL
GROUP BY - correct and simple at today's scale; if usage_events grows
into the millions (plausible, given retention is indefinite by decision),
the day/week/month bucketing here is the first thing to move server-side
into SQL. Not done now because nothing currently requires it."""
rows = db.scalars(_usage_query(db, caller, date_from, date_to, project_id, username, tool)).all()
by_day: dict[str, set] = {}
by_week: dict[str, set] = {}
by_month: dict[str, set] = {}
per_user_last: dict[str, datetime] = {}
per_tool: dict[str, int] = {}
for e in rows:
by_day.setdefault(e.at.date().isoformat(), set()).add(e.username)
by_week.setdefault(e.at.strftime("%G-W%V"), set()).add(e.username)
by_month.setdefault(e.at.strftime("%Y-%m"), set()).add(e.username)
if e.username not in per_user_last or e.at > per_user_last[e.username]:
per_user_last[e.username] = e.at
per_tool[e.tool] = per_tool.get(e.tool, 0) + 1
return {
"active_users": {
"by_day": {k: len(v) for k, v in sorted(by_day.items())},
"by_week": {k: len(v) for k, v in sorted(by_week.items())},
"by_month": {k: len(v) for k, v in sorted(by_month.items())},
},
"per_user_last_active": {u: models._iso(t) for u, t in sorted(per_user_last.items())},
"by_tool": dict(sorted(per_tool.items(), key=lambda kv: -kv[1])),
"event_count": len(rows),
}
@app.get("/api/comments")
def list_comments(
source: Optional[str] = Query(None),