curo1305
3ec198768d
- users.py: list_users, create_user, update_user_status, initiate_password_reset, update_ai_config, delete_user, create_system_topic (7 routes) - quotas.py: get_user_quota, update_user_quota (2 routes) - ai.py: get_ai_config_models, test_ai_connection, get_ai_config, update_system_ai_config (4 routes); validate_provider_id call-through (D-11) - shared.py: _user_to_dict helper (T-02-27 / SEC-07 field whitelist) - __init__.py: router aggregator with prefix=/api/admin; 13 routes total - admin.py monolith renamed to admin_OLD_REMOVE_IN_TASK_3.py (deleted in task 3) - All sub-routers have NO prefix (D-04) - All handlers inject Depends(get_current_admin) (T-08-04-01)
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""Admin quota management endpoints.
|
|
|
|
Handles: get_user_quota, update_user_quota.
|
|
|
|
All handlers require get_current_admin (SEC-07, T-08-04-01).
|
|
Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel, field_validator
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models import Quota, User
|
|
from deps.auth import get_current_admin
|
|
from deps.db import get_db
|
|
from deps.utils import get_client_ip
|
|
from services.audit import write_audit_log
|
|
from api.admin.shared import _user_to_dict
|
|
|
|
router = APIRouter() # NO prefix — parent __init__.py carries /api/admin (D-04)
|
|
|
|
|
|
# ── Request models ────────────────────────────────────────────────────────────
|
|
|
|
class QuotaUpdate(BaseModel):
|
|
limit_bytes: int
|
|
|
|
@field_validator("limit_bytes")
|
|
@classmethod
|
|
def must_be_positive(cls, v: int) -> int:
|
|
if v <= 0:
|
|
raise ValueError("limit_bytes must be greater than 0")
|
|
return v
|
|
|
|
|
|
# ── Endpoints ─────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/users/{user_id}/quota")
|
|
async def get_user_quota(
|
|
user_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_db),
|
|
_admin: User = Depends(get_current_admin),
|
|
) -> dict:
|
|
"""Return quota details for a user (ADMIN-04).
|
|
|
|
Quota info is admin-visible operational data — no PII, no document content
|
|
(T-02-31 disposition: accept).
|
|
"""
|
|
quota = await session.get(Quota, user_id)
|
|
if quota is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
|
|
|
|
return {
|
|
"user_id": str(quota.user_id),
|
|
"limit_bytes": quota.limit_bytes,
|
|
"used_bytes": quota.used_bytes,
|
|
"limit_mb": quota.limit_bytes // 1048576,
|
|
"used_mb": quota.used_bytes // 1048576,
|
|
}
|
|
|
|
|
|
@router.patch("/users/{user_id}/quota")
|
|
async def update_user_quota(
|
|
user_id: uuid.UUID,
|
|
body: QuotaUpdate,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
_admin: User = Depends(get_current_admin),
|
|
) -> dict:
|
|
"""Adjust a user's storage quota (ADMIN-04).
|
|
|
|
If the new limit is below current usage, still applies the change but
|
|
returns warning=True with an explanatory message. Uploads will be blocked
|
|
but existing documents are preserved.
|
|
"""
|
|
quota = await session.get(Quota, user_id)
|
|
if quota is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
|
|
|
|
warning = body.limit_bytes < quota.used_bytes
|
|
warning_message = (
|
|
"New limit is below current usage. Uploads will be blocked but existing documents are preserved."
|
|
if warning
|
|
else None
|
|
)
|
|
|
|
_ip_addr = get_client_ip(request)
|
|
old_limit = quota.limit_bytes
|
|
quota.limit_bytes = body.limit_bytes
|
|
session.add(quota)
|
|
|
|
# D-13: quota changed event
|
|
await write_audit_log(
|
|
session,
|
|
event_type="admin.quota_changed",
|
|
user_id=user_id,
|
|
actor_id=_admin.id,
|
|
resource_id=None,
|
|
ip_address=_ip_addr,
|
|
metadata_={"old_bytes": old_limit, "new_bytes": body.limit_bytes},
|
|
)
|
|
await session.commit()
|
|
|
|
response: dict = {
|
|
"user_id": str(quota.user_id),
|
|
"limit_bytes": quota.limit_bytes,
|
|
"used_bytes": quota.used_bytes,
|
|
"warning": warning,
|
|
}
|
|
if warning_message:
|
|
response["message"] = warning_message
|
|
return response
|