feat(quotas): admin quota API and GET /api/user/quota

Admins read and set the instance default, team allowances and user
overrides under /api/admin/quotas. A user's endpoint also returns the
limits those layers resolve to, the layer each came from and the usage
against them. The overview lists catalog models used this period that have
no price, since a cost limit cannot see them. Every write is audited.

GET /api/user/quota gives a user their own limited buckets, usage and reset
time without naming the policies behind them; any valid token may call it.
This commit is contained in:
Alex committed 2026-09-21 12:11:12 +01:00
1 parent 5406550bca
commit 21f43b2966
6 files changed
+616

No files matched your search

+36
View File
@@ -5,6 +5,9 @@ only from ``request.decoded_token`` (already populated and role-resolved by the
auth chokepoint in ``app.py``). Auth-mode-agnostic. ``email``/``name``/
``picture`` are OIDC-only and optional — they are echoed from the token and are
never present for ``simple_jwt``/``session_jwt``/no-auth modes.
``GET /api/user/quota`` returns the caller's usage against the limits an admin
set for them, without naming the policies behind those limits.
"""
from __future__ import annotations
@@ -13,6 +16,8 @@ from flask import jsonify, make_response, request
from flask_restx import Namespace, Resource
from docsgpt.api.pat.tokens import is_pat
from docsgpt.core.settings import settings
from docsgpt.quotas.service import REQUEST_BUCKETS, QuotaService
me_ns = Namespace("me", description="Current user identity and roles", path="/api")
@@ -43,3 +48,34 @@ class MeResource(Resource):
"resource_filter": decoded_token.get("resource_filter") or {},
}
return make_response(jsonify(body), 200)
def _own_budget(budget: dict) -> dict:
return {"limit": budget["limit"], "used": budget["used"]}
@me_ns.route("/user/quota")
class MyQuotaResource(Resource):
def get(self):
"""Return the caller's limited buckets: ``{bucket, tokens, cost, resets_at}`` each."""
decoded_token = getattr(request, "decoded_token", None)
user_id = decoded_token.get("sub") if decoded_token else None
if not user_id:
return make_response(jsonify({"success": False}), 401)
statuses = QuotaService.status(user_id, ("all", *REQUEST_BUCKETS))
buckets = []
for status in statuses:
if status.limits.unlimited:
continue
data = status.to_dict()
buckets.append(
{
"bucket": data["bucket"],
"tokens": _own_budget(data["tokens"]),
"cost": _own_budget(data["cost"]),
"resets_at": data["resets_at"],
}
)
return make_response(
jsonify({"success": True, "period": settings.QUOTA_PERIOD, "buckets": buckets}), 200
)