retoorandClaude Sonnet 5
57087536e5
- Route Devii-driven AI gateway cost to the action/tool that triggered it instead of a blanket "internal" bucket, so per-feature AI spend is attributable. - Fix the quiz attempt review to show one previously-answered question at a time instead of all of them at once, and stop a quiz endpoint linked from the quiz flow from responding with raw JSON. - Add DB API async query result route and AI Usage Analyzer annotated source/media routes, with traversal-safe uid/path handling and matching tests. - Add Code Farm action audit logging (plant/harvest/buy-plot/upgrade/ fertilize) and related admin workspace/services/trash/gateway route and doc touch-ups. - Drop redundant docstrings from access_tokens.py per the no-comments convention. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VL9Xn57W5UR3HZbbuuzxdK
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
import json
|
|
import logging
|
|
from typing import Any, TypeVar, get_origin
|
|
|
|
from fastapi import HTTPException, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from pydantic import BaseModel, ValidationError
|
|
from starlette.datastructures import FormData
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_TModel = TypeVar("_TModel", bound=BaseModel)
|
|
|
|
_SEQUENCE_ORIGINS = frozenset({list, set, tuple, frozenset})
|
|
|
|
|
|
def _formdata_to_dict(form: FormData, model: type[BaseModel]) -> dict[str, Any]:
|
|
body: dict[str, Any] = {}
|
|
for field_name, field_info in model.model_fields.items():
|
|
origin = get_origin(field_info.annotation)
|
|
if origin in _SEQUENCE_ORIGINS:
|
|
values = form.getlist(field_name)
|
|
if not values:
|
|
continue
|
|
if values == [""]:
|
|
continue
|
|
body[field_name] = [v for v in values if v != ""] or []
|
|
else:
|
|
value = form.get(field_name)
|
|
if value is not None:
|
|
body[field_name] = value
|
|
return body
|
|
|
|
|
|
class _JsonOrForm:
|
|
def __init__(self, model: type[BaseModel]):
|
|
self.model = model
|
|
|
|
async def __call__(self, request: Request) -> Any:
|
|
content_type = request.headers.get("content-type", "")
|
|
body: Any = None
|
|
try:
|
|
if "application/json" in content_type:
|
|
try:
|
|
body = await request.json()
|
|
except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc:
|
|
logger.debug("JSON parse failed: %s", exc)
|
|
raise HTTPException(status_code=400, detail="Invalid JSON body")
|
|
if not isinstance(body, dict):
|
|
raise HTTPException(
|
|
status_code=400, detail="JSON body must be an object"
|
|
)
|
|
return self.model.model_validate(body)
|
|
try:
|
|
form = await request.form()
|
|
except Exception as exc:
|
|
logger.debug("Form parse failed: %s", exc)
|
|
raise HTTPException(
|
|
status_code=400, detail="Could not parse form data"
|
|
)
|
|
body = _formdata_to_dict(form, self.model)
|
|
return self.model.model_validate(body)
|
|
except ValidationError as exc:
|
|
raise RequestValidationError(errors=exc.errors(), body=body)
|
|
|
|
|
|
def json_or_form(model: type[_TModel]) -> _JsonOrForm:
|
|
return _JsonOrForm(model)
|