curo1305
6d1c02f703
- api/admin/users.py: removed module docstring + 7 WHAT function docstrings - api/admin/quotas.py: removed module docstring + 2 WHAT function docstrings - api/admin/ai.py: removed module docstring + 3 WHAT inline comments; security invariant docstrings preserved - api/admin/shared.py: unchanged (all comments are WHY — constraint notes) - api/documents/upload.py: removed 4 WHAT inline comments; T-03-05/T-03-06 WHY notes preserved - api/documents/content.py: removed WHAT function docstring + WHAT inline comment - api/documents/crud.py: removed 7 WHAT inline comments; D-16 + security constraint docstrings preserved - api/auth/shared.py: removed module docstring + 1 WHAT inline comment - api/auth/tokens.py: removed module docstring + 8 WHAT function docstrings/comments; family-revocation + SEC-02 WHY notes preserved - api/auth/totp.py: removed module docstring + 2 WHAT function docstrings + 2 WHAT inline comments - api/auth/password.py: removed module docstring + 2 WHAT function docstrings + 3 WHAT inline comments - All NO-prefix anchor comments and security-invariant WHY comments preserved - pytest: 413 passed, 1 failed (pre-existing ModuleNotFoundError unrelated to purge)
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
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:
|
|
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:
|
|
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
|