39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
"""Read-only activity log. Admin-only: shows every state-changing action and
|
|
auth event, filterable by acting user."""
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from ..auth import require_admin
|
|
from ..database import get_db
|
|
from ..models import Admin, LogEntry
|
|
from ..pagination import Page, PageParams
|
|
from ..schemas import LogOut
|
|
|
|
router = APIRouter(prefix="/logs", tags=["logs"])
|
|
|
|
|
|
@router.get("", response_model=Page[LogOut])
|
|
def list_logs(
|
|
page: PageParams = Depends(),
|
|
username: str | None = None,
|
|
action: str | None = None,
|
|
_: Admin = Depends(require_admin),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
stmt = select(LogEntry)
|
|
count_stmt = select(func.count()).select_from(LogEntry)
|
|
if username:
|
|
stmt = stmt.where(LogEntry.username == username)
|
|
count_stmt = count_stmt.where(LogEntry.username == username)
|
|
if action:
|
|
stmt = stmt.where(LogEntry.action == action)
|
|
count_stmt = count_stmt.where(LogEntry.action == action)
|
|
|
|
total = int(db.execute(count_stmt).scalar_one())
|
|
rows = db.execute(
|
|
stmt.order_by(LogEntry.id.desc()).limit(page.limit).offset(page.offset)
|
|
).scalars().all()
|
|
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
|