counterweight
6c218130e9
- Add ruff as dev dependency - Configure ruff in pyproject.toml with strict 88-char line limit - Ignore B008 (FastAPI Depends pattern is standard) - Allow longer lines in tests for readability - Fix all lint issues in source files - Add Makefile targets: lint-backend, format-backend, fix-backend
117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
"""Audit routes for viewing action records."""
|
|
|
|
from collections.abc import Callable
|
|
from typing import TypeVar
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import desc, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from auth import require_permission
|
|
from database import get_db
|
|
from models import CounterRecord, Permission, SumRecord, User
|
|
from schemas import (
|
|
CounterRecordResponse,
|
|
PaginatedCounterRecords,
|
|
PaginatedSumRecords,
|
|
SumRecordResponse,
|
|
)
|
|
|
|
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,
|
|
)
|