# retoor <retoor@molodetz.nl>
import logging
from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import ValidationError
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.services.audit import record as audit
from devplacepy.services.manager import service_manager
from devplacepy.services.openai_gateway import quota, routing
from devplacepy.templating import templates
from devplacepy.utils import require_admin
logger = logging.getLogger(__name__)
router = APIRouter()
def _default_provider_summary() -> dict:
svc = service_manager.get_service("openai")
cfg = svc.get_config() if svc is not None else {}
return {
"base_url": cfg.get("gateway_upstream_url", ""),
"model": cfg.get("gateway_model", ""),
"embed_url": cfg.get("gateway_embed_url", ""),
"embed_model": cfg.get("gateway_embed_model", ""),
"image_url": cfg.get("gateway_image_url", ""),
"image_model": cfg.get("gateway_image_model", ""),
"vision_url": cfg.get("gateway_vision_url", ""),
"vision_model": cfg.get("gateway_vision_model", ""),
}
def _validation_error(exc: ValidationError) -> JSONResponse:
first = exc.errors()[0]
message = first.get("msg", "Invalid input")
return JSONResponse({"ok": False, "error": message}, status_code=400)
async def _payload(request: Request) -> dict:
content_type = request.headers.get("content-type", "")
if "application/json" in content_type:
try:
data = await request.json()
except Exception:
return {}
return data if isinstance(data, dict) else {}
form = await request.form()
return {key: value for key, value in form.items()}
@router.get("/gateway", response_class=HTMLResponse)
async def gateway_config_page(request: Request):
admin = require_admin(request)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Gateway routing - Admin",
description="Manage OpenAI gateway providers and per-model routing.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Gateway", "url": "/admin/gateway"},
],
schemas=[website_schema(base)],
)
return templates.TemplateResponse(
request,
"admin_gateway.html",
{
**seo_ctx,
"request": request,
"user": admin,
"admin_section": "gateway",
"default_provider": _default_provider_summary(),
},
)
@router.get("/gateway/providers")
async def list_providers(request: Request):
require_admin(request)
return JSONResponse(
{
"providers": routing.provider_store.list(),
"default": _default_provider_summary(),
"count": routing.provider_store.count(),
}
)
@router.post("/gateway/providers")
async def save_provider(request: Request):
admin = require_admin(request)
body = await _payload(request)
try:
payload = routing.ProviderIn(**body)
except ValidationError as exc:
return _validation_error(exc)
saved = routing.provider_store.set(payload)
audit.record(
request,
"gateway.provider.update",
user=admin,
target_type="gateway_provider",
target_uid=payload.name,
target_label=payload.name,
summary=f"admin {admin['username']} saved gateway provider {payload.name}",
)
return JSONResponse({"ok": True, "provider": saved})
@router.delete("/gateway/providers/{name}")
async def delete_provider(request: Request, name: str):
admin = require_admin(request)
existed = routing.provider_store.remove(name)
if not existed:
return JSONResponse({"ok": False, "error": "Provider not found"}, status_code=404)
audit.record(
request,
"gateway.provider.delete",
user=admin,
target_type="gateway_provider",
target_uid=name,
target_label=name,
summary=f"admin {admin['username']} deleted gateway provider {name}",
)
return JSONResponse({"ok": True})
@router.get("/gateway/models")
async def list_models(request: Request):
require_admin(request)
return JSONResponse(
{
"models": routing.model_store.list(),
"providers": routing.provider_store.names(),
"count": routing.model_store.count(),
}
)
@router.post("/gateway/models")
async def save_model(request: Request):
admin = require_admin(request)
body = await _payload(request)
try:
payload = routing.ModelRouteIn(**body)
except ValidationError as exc:
return _validation_error(exc)
saved = routing.model_store.set(payload)
audit.record(
request,
"gateway.model.update",
user=admin,
target_type="gateway_model",
target_uid=payload.source_model,
target_label=f"{payload.source_model} -> {payload.target_model}",
summary=(
f"admin {admin['username']} saved gateway model route "
f"{payload.source_model} -> {payload.target_model}"
),
)
return JSONResponse({"ok": True, "model": saved})
@router.delete("/gateway/models/{source_model}")
async def delete_model(request: Request, source_model: str):
admin = require_admin(request)
existed = routing.model_store.remove(source_model)
if not existed:
return JSONResponse({"ok": False, "error": "Model route not found"}, status_code=404)
audit.record(
request,
"gateway.model.delete",
user=admin,
target_type="gateway_model",
target_uid=source_model,
target_label=source_model,
summary=f"admin {admin['username']} deleted gateway model route {source_model}",
)
return JSONResponse({"ok": True})
def _quota_defaults_summary() -> dict:
svc = service_manager.get_service("openai")
cfg = svc.get_config() if svc is not None else {}
return {
"user": cfg.get(quota.FIELD_DEFAULT_USER, 0.0),
"admin": cfg.get(quota.FIELD_DEFAULT_ADMIN, 0.0),
"guest": cfg.get(quota.FIELD_DEFAULT_GUEST, 0.0),
"internal": cfg.get(quota.FIELD_DEFAULT_INTERNAL, 0.0),
"key": cfg.get(quota.FIELD_DEFAULT_KEY, 0.0),
}
def _rule_label(rule: dict) -> str:
return quota.scope_label(rule, fallback=rule.get("uid", ""))
@router.get("/gateway/quota-rules")
async def list_quota_rules(request: Request):
require_admin(request)
rules = quota.quota_rule_store.list()
for rule in rules:
rule["spent_24h_usd"] = round(
quota.spent_24h(rule["owner_kind"], rule["owner_id"], rule["app_reference"]), 6
)
return JSONResponse(
{
"rules": rules,
"count": len(rules),
"defaults": _quota_defaults_summary(),
}
)
@router.post("/gateway/quota-rules")
async def save_quota_rule(request: Request):
admin = require_admin(request)
body = await _payload(request)
uid = str(body.pop("uid", "") or "").strip() or None
try:
payload = quota.QuotaRuleIn(**body)
except ValidationError as exc:
return _validation_error(exc)
saved = quota.quota_rule_store.set(payload, uid=uid, created_by=admin["uid"])
audit.record(
request,
"gateway.quota_rule.update",
user=admin,
target_type="gateway_quota_rule",
target_uid=saved["uid"],
target_label=_rule_label(saved),
summary=f"admin {admin['username']} saved gateway quota rule ({_rule_label(saved)}) at ${saved['limit_usd']}/24h",
metadata={
"owner_kind": saved["owner_kind"],
"owner_id": saved["owner_id"],
"app_reference": saved["app_reference"],
"limit_usd": saved["limit_usd"],
"is_active": saved["is_active"],
},
)
return JSONResponse({"ok": True, "rule": saved})
@router.post("/gateway/quota-resets")
async def reset_quota_spend(request: Request):
admin = require_admin(request)
body = await _payload(request)
try:
payload = quota.QuotaResetIn(**body)
except ValidationError as exc:
return _validation_error(exc)
scope = quota.reset(payload, created_by=admin["uid"])
label = quota.scope_label(scope, fallback="every caller")
audit.record(
request,
"gateway.quota.reset",
user=admin,
target_type="gateway_quota",
target_uid=scope["uid"],
target_label=label,
summary=f"admin {admin['username']} reset the gateway 24h spend for {label}",
metadata={
"owner_kind": scope["owner_kind"],
"owner_id": scope["owner_id"],
"app_reference": scope["app_reference"],
"reset_at": scope["reset_at"],
},
)
return JSONResponse({"ok": True, "reset": scope})
@router.delete("/gateway/quota-rules/{uid}")
async def delete_quota_rule(request: Request, uid: str):
admin = require_admin(request)
existing = quota.quota_rule_store.get(uid)
label = _rule_label(existing.as_dict()) if existing else uid
existed = quota.quota_rule_store.remove(uid)
if not existed:
return JSONResponse({"ok": False, "error": "Quota rule not found"}, status_code=404)
audit.record(
request,
"gateway.quota_rule.delete",
user=admin,
target_type="gateway_quota_rule",
target_uid=uid,
target_label=label,
summary=f"admin {admin['username']} deleted gateway quota rule ({label})",
)
return JSONResponse({"ok": True})
View Git Blame Copy Permalink