first implementation

backend/routes/audit.py

@@ -0,0 +1,117 @@
+"""Audit routes for viewing action records."""
+from typing import Callable, TypeVar
+
+from fastapi import APIRouter, Depends, Query
+from pydantic import BaseModel
+from sqlalchemy import select, func, desc
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from auth import require_permission
+from database import get_db
+from models import User, SumRecord, CounterRecord, Permission
+from schemas import (
+    CounterRecordResponse,
+    SumRecordResponse,
+    PaginatedCounterRecords,
+    PaginatedSumRecords,
+)
+
+
+router = APIRouter(prefix="/api/audit", tags=["audit"])
+
+R = TypeVar("R", bound=BaseModel)
+
+
+async def paginate_with_user_email(
+    db: AsyncSession,
+    model: type[SumRecord] | type[CounterRecord],
+    page: int,
+    per_page: int,
+    row_mapper: Callable[..., R],
+) -> tuple[list[R], int, int]:
+    """
+    Generic pagination helper for audit records that need user email.
+
+    Returns: (records, total, total_pages)
+    """
+    # Get total count
+    count_result = await db.execute(select(func.count(model.id)))
+    total = count_result.scalar() or 0
+    total_pages = (total + per_page - 1) // per_page if total > 0 else 1
+
+    # Get paginated records with user email
+    offset = (page - 1) * per_page
+    query = (
+        select(model, User.email)
+        .join(User, model.user_id == User.id)
+        .order_by(desc(model.created_at))
+        .offset(offset)
+        .limit(per_page)
+    )
+    result = await db.execute(query)
+    rows = result.all()
+
+    records: list[R] = [row_mapper(record, email) for record, email in rows]
+    return records, total, total_pages
+
+
+def _map_counter_record(record: CounterRecord, email: str) -> CounterRecordResponse:
+    return CounterRecordResponse(
+        id=record.id,
+        user_email=email,
+        value_before=record.value_before,
+        value_after=record.value_after,
+        created_at=record.created_at,
+    )
+
+
+def _map_sum_record(record: SumRecord, email: str) -> SumRecordResponse:
+    return SumRecordResponse(
+        id=record.id,
+        user_email=email,
+        a=record.a,
+        b=record.b,
+        result=record.result,
+        created_at=record.created_at,
+    )
+
+
+@router.get("/counter", response_model=PaginatedCounterRecords)
+async def get_counter_records(
+    page: int = Query(1, ge=1),
+    per_page: int = Query(10, ge=1, le=100),
+    db: AsyncSession = Depends(get_db),
+    _current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)),
+) -> PaginatedCounterRecords:
+    """Get paginated counter action records."""
+    records, total, total_pages = await paginate_with_user_email(
+        db, CounterRecord, page, per_page, _map_counter_record
+    )
+    return PaginatedCounterRecords(
+        records=records,
+        total=total,
+        page=page,
+        per_page=per_page,
+        total_pages=total_pages,
+    )
+
+
+@router.get("/sum", response_model=PaginatedSumRecords)
+async def get_sum_records(
+    page: int = Query(1, ge=1),
+    per_page: int = Query(10, ge=1, le=100),
+    db: AsyncSession = Depends(get_db),
+    _current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)),
+) -> PaginatedSumRecords:
+    """Get paginated sum action records."""
+    records, total, total_pages = await paginate_with_user_email(
+        db, SumRecord, page, per_page, _map_sum_record
+    )
+    return PaginatedSumRecords(
+        records=records,
+        total=total,
+        page=page,
+        per_page=per_page,
+        total_pages=total_pages,
+    )
+