agent-platform/services/api-gateway/app/infrastructure/audit.py

from time import perf_counter

from fastapi import Request
from sqlalchemy.orm import sessionmaker

from app.domain.repositories import GatewayRequestAuditRepository
from app.infrastructure.request_context import get_gateway_request_context


def mark_gateway_target(
    request: Request,
    *,
    target_service: str,
    target_url: str,
) -> None:
    context = get_gateway_request_context(request)
    context.target_service = target_service
    context.target_url = target_url


def persist_gateway_audit(
    *,
    request: Request,
    session_factory: sessionmaker,
    status_code: int | None,
    error_message: str | None = None,
) -> None:
    context = get_gateway_request_context(request)
    duration_ms = int((perf_counter() - context.started_perf_counter) * 1000)
    client_host = request.client.host if request.client is not None else None
    query_string = request.url.query or None
    user_agent = request.headers.get("user-agent")

    db = session_factory()
    try:
        GatewayRequestAuditRepository(db).create(
            tenant_id=context.tenant_id,
            request_id=context.request_id,
            method=request.method,
            path=request.url.path,
            query_string=query_string,
            target_service=context.target_service,
            target_url=context.target_url,
            status_code=status_code,
            duration_ms=duration_ms,
            client_host=client_host,
            user_agent=user_agent,
            error_message=error_message,
        )
    finally:
        db.close()