| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960 |
- from sqlalchemy import select
- from sqlalchemy.orm import Session
- from app.db.models import GatewayRequestAudit
- class GatewayRequestAuditRepository:
- def __init__(self, db: Session) -> None:
- self.db = db
- def create(
- self,
- *,
- tenant_id: str,
- request_id: str,
- method: str,
- path: str,
- query_string: str | None,
- target_service: str | None,
- target_url: str | None,
- status_code: int | None,
- duration_ms: int,
- client_host: str | None,
- user_agent: str | None,
- error_message: str | None,
- ) -> GatewayRequestAudit:
- entity = GatewayRequestAudit(
- tenant_id=tenant_id,
- request_id=request_id,
- method=method,
- path=path,
- query_string=query_string,
- target_service=target_service,
- target_url=target_url,
- status_code=status_code,
- duration_ms=duration_ms,
- client_host=client_host,
- user_agent=user_agent,
- error_message=error_message,
- )
- self.db.add(entity)
- self.db.commit()
- self.db.refresh(entity)
- return entity
- def list_by_scope(
- self,
- *,
- tenant_id: str,
- request_id: str | None = None,
- target_service: str | None = None,
- limit: int = 100,
- ) -> list[GatewayRequestAudit]:
- stmt = select(GatewayRequestAudit).where(GatewayRequestAudit.tenant_id == tenant_id)
- if request_id is not None:
- stmt = stmt.where(GatewayRequestAudit.request_id == request_id)
- if target_service is not None:
- stmt = stmt.where(GatewayRequestAudit.target_service == target_service)
- stmt = stmt.order_by(GatewayRequestAudit.created_time.desc()).limit(limit)
- return list(self.db.scalars(stmt))
|
