93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
"""Layered settings: team defaults under per-user overrides, key by key."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.responses import JSONResponse, Response
|
|
from pydantic import BaseModel
|
|
|
|
from ..auth import Ctx
|
|
from ..db import record_audit
|
|
from ..deps import get_ctx
|
|
from ..util import SETTING_KEY_RE, sha256_text, now_iso
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SettingWrite(BaseModel):
|
|
value: Any
|
|
base_version: int | None = None
|
|
|
|
|
|
@router.get("/v1/settings")
|
|
def get_settings(request: Request, ctx: Ctx = Depends(get_ctx)):
|
|
user = ctx.require_user(request)
|
|
rows = ctx.conn.execute(
|
|
"""SELECT scope, key, value_json, version, updated_at FROM settings
|
|
WHERE scope = 'team' OR (scope = 'user' AND scope_id = %s)
|
|
ORDER BY CASE WHEN scope = 'user' THEN 1 ELSE 0 END""",
|
|
(user["id"],),
|
|
).fetchall()
|
|
settings: dict[str, dict] = {}
|
|
for r in rows: # user rows are ordered last, so they win per key
|
|
settings[r["key"]] = {
|
|
"value": json.loads(r["value_json"]),
|
|
"version": r["version"],
|
|
"scope": r["scope"],
|
|
"updated_at": r["updated_at"],
|
|
}
|
|
payload = {"settings": settings}
|
|
etag = '"' + sha256_text(json.dumps(payload, sort_keys=True, separators=(",", ":")))[:32] + '"'
|
|
inm = request.headers.get("if-none-match")
|
|
if inm and etag in [t.strip() for t in inm.split(",")]:
|
|
return Response(status_code=304, headers={"ETag": etag})
|
|
return JSONResponse(payload, headers={"ETag": etag})
|
|
|
|
|
|
def _put_setting(request: Request, ctx: Ctx, scope: str, key: str, body: SettingWrite):
|
|
user = ctx.require_user(request)
|
|
if scope == "team" and user["role"] != "admin":
|
|
raise HTTPException(403, "team settings are admin-only")
|
|
if not SETTING_KEY_RE.match(key):
|
|
raise HTTPException(400, "invalid setting key")
|
|
scope_id = 0 if scope == "team" else user["id"]
|
|
row = ctx.conn.execute(
|
|
"SELECT version FROM settings WHERE scope = %s AND scope_id = %s AND key = %s",
|
|
(scope, scope_id, key),
|
|
).fetchone()
|
|
if body.base_version is not None:
|
|
current = row["version"] if row else 0
|
|
if current != body.base_version:
|
|
raise HTTPException(
|
|
409,
|
|
{"message": "version conflict", "current_version": current},
|
|
)
|
|
new_version = (row["version"] + 1) if row else 1
|
|
ts = now_iso()
|
|
ctx.conn.execute(
|
|
"""INSERT INTO settings (scope, scope_id, key, value_json, version, updated_by, updated_at)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
ON CONFLICT (scope, scope_id, key) DO UPDATE SET
|
|
value_json = excluded.value_json,
|
|
version = excluded.version,
|
|
updated_by = excluded.updated_by,
|
|
updated_at = excluded.updated_at""",
|
|
(scope, scope_id, key, json.dumps(body.value), new_version, user["id"], ts),
|
|
)
|
|
ctx.conn.commit()
|
|
if scope == "team":
|
|
record_audit(ctx.conn, user["id"], "settings.team.update", {"key": key, "version": new_version})
|
|
return {"key": key, "scope": scope, "version": new_version, "updated_at": ts}
|
|
|
|
|
|
@router.put("/v1/settings/user/{key}")
|
|
def put_user_setting(key: str, body: SettingWrite, request: Request, ctx: Ctx = Depends(get_ctx)):
|
|
return _put_setting(request, ctx, "user", key, body)
|
|
|
|
|
|
@router.put("/v1/settings/team/{key}")
|
|
def put_team_setting(key: str, body: SettingWrite, request: Request, ctx: Ctx = Depends(get_ctx)):
|
|
return _put_setting(request, ctx, "team", key, body)
|