9960d760a0
- 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>
82 lines
2.2 KiB
Python
82 lines
2.2 KiB
Python
"""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,
|
|
})
|