| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647 |
- 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(
- 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()
|
