feat: API request/response logging with admin log viewer

- Add api_logs table (migration 0007) and ApiLog model
- Add web/lib/api_logger.py — httpx wrapper that records every outbound call
- Wire api_logger into vk_sync, vk_catalog, and connections test endpoints
- Add /admin/logs page with filters (service, method, status, time range, URL search) and expandable request/response detail
- Add "Логи" nav link for admin users

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mguschin
2026-05-12 22:00:14 +03:00
parent cad0b10fbb
commit 9960d760a0
10 changed files with 392 additions and 21 deletions

View File

@@ -3,6 +3,7 @@ from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode
import httpx
import web.lib.api_logger as api_logger
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from sqlalchemy.orm import Session
@@ -306,8 +307,9 @@ async def connections_evotor_test(request: Request, db: Session = Depends(get_db
return JSONResponse({"ok": False, "message": "Подключение не настроено"})
try:
r = httpx.get(
r = api_logger.get(
"https://api.evotor.ru/stores",
user_id=user.id,
headers={
"Authorization": f"Bearer {conn.access_token}",
"Accept": "application/vnd.evotor.v2+json",
@@ -344,8 +346,9 @@ async def connections_vk_test(request: Request, db: Session = Depends(get_db)):
if not conn.vk_user_id:
return JSONResponse({"ok": False, "message": "Укажите ID сообщества для проверки подключения."})
r = httpx.get(
r = api_logger.get(
"https://api.vk.com/method/groups.getById",
user_id=user.id,
params={
"group_id": conn.vk_user_id,
"fields": "market",

81
web/routes/logs.py Normal file
View File

@@ -0,0 +1,81 @@
"""API request/response log viewer (admin only)."""
from datetime import datetime, timedelta
from fastapi import APIRouter, Depends, Request
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from web.auth.session import get_current_user
from web.database import get_db
from web.models.connections import ApiLog
from web.templates_env import templates
router = APIRouter()
PAGE_SIZE = 50
def _render(request, template, ctx):
return templates.TemplateResponse(template, {"request": request, **ctx})
@router.get("/admin/logs")
async def admin_logs(
request: Request,
db: Session = Depends(get_db),
service: str = "",
method: str = "",
status: str = "",
q: str = "",
page: int = 1,
hours: int = 168,
):
try:
user = get_current_user(request, db)
except Exception:
return RedirectResponse("/login", 303)
since = datetime.utcnow() - timedelta(hours=hours)
query = db.query(ApiLog).filter(ApiLog.created_at >= since)
if service:
query = query.filter(ApiLog.service == service)
if method:
query = query.filter(ApiLog.method == method)
if status:
try:
st = int(status)
query = query.filter(ApiLog.response_status == st)
except ValueError:
if status == "error":
query = query.filter(ApiLog.response_status >= 400)
elif status == "ok":
query = query.filter(ApiLog.response_status < 400)
if q:
like = f"%{q}%"
query = query.filter(
ApiLog.url.like(like) | ApiLog.response_body.like(like)
)
total = query.count()
logs = (
query.order_by(ApiLog.created_at.desc())
.offset((page - 1) * PAGE_SIZE)
.limit(PAGE_SIZE)
.all()
)
total_pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
return _render(request, "admin/logs.html", {
"user": user,
"logs": logs,
"total": total,
"page": page,
"total_pages": total_pages,
"page_size": PAGE_SIZE,
"filter_service": service,
"filter_method": method,
"filter_status": status,
"filter_q": q,
"filter_hours": hours,
})