| 1234567891011121314151617181920212223242526272829303132333435363738394041 |
- from __future__ import annotations
- from datetime import datetime
- from typing import List, Optional
- from fastapi import APIRouter, Query
- from pydantic import BaseModel
- from app.db import get_pool
- router = APIRouter(tags=["logs"])
- class AccessLogOut(BaseModel):
- id: int
- ip: str
- method: str
- path: str
- status_code: int
- latency_ms: float
- country: str
- city: str
- latitude: Optional[float]
- longitude: Optional[float]
- created_at: datetime
- @router.get("/logs", response_model=List[AccessLogOut])
- async def get_logs(
- page: int = Query(default=1, ge=1),
- page_size: int = Query(default=50, ge=1, le=500),
- ) -> List[AccessLogOut]:
- offset = (page - 1) * page_size
- pool = get_pool()
- async with pool.acquire() as conn:
- rows = await conn.fetch(
- "SELECT * FROM access_logs ORDER BY created_at DESC LIMIT $1 OFFSET $2",
- page_size,
- offset,
- )
- return [AccessLogOut(**dict(row)) for row in rows]
|
