""" API audit log model for tracking API requests and security events. Tracks all API requests including user, endpoint, request/response details, and performance metrics for security auditing and monitoring. """ from datetime import datetime from typing import Optional from sqlalchemy import Index, Integer, String, Text, TIMESTAMP from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from .base import Base, UUIDMixin class ApiAuditLog(Base, UUIDMixin): """ API audit log model for tracking API requests and security. Logs all API requests with details about the user, endpoint accessed, request/response data, performance metrics, and errors. Used for security auditing, monitoring, and troubleshooting API issues. Attributes: user_id: User identifier from JWT sub claim endpoint: API endpoint path accessed http_method: HTTP method used (GET, POST, PUT, DELETE, etc.) ip_address: IP address of the requester user_agent: User agent string from the request request_body: Sanitized request body (credentials removed) response_status: HTTP response status code response_time_ms: Response time in milliseconds error_message: Error message if request failed timestamp: When the request was made """ __tablename__ = "api_audit_log" # User identification user_id: Mapped[str] = mapped_column( String(255), nullable=False, doc="User identifier from JWT sub claim" ) # Request details endpoint: Mapped[str] = mapped_column( String(500), nullable=False, doc="API endpoint path accessed (e.g., '/api/v1/sessions')" ) http_method: Mapped[Optional[str]] = mapped_column( String(10), doc="HTTP method used: GET, POST, PUT, DELETE, PATCH" ) # Client information ip_address: Mapped[Optional[str]] = mapped_column( String(45), doc="IP address of the requester (IPv4 or IPv6)" ) user_agent: Mapped[Optional[str]] = mapped_column( Text, doc="User agent string from the request" ) # Request/Response data request_body: Mapped[Optional[str]] = mapped_column( Text, doc="Sanitized request body (credentials and sensitive data removed)" ) response_status: Mapped[Optional[int]] = mapped_column( Integer, doc="HTTP response status code (200, 401, 500, etc.)" ) response_time_ms: Mapped[Optional[int]] = mapped_column( Integer, doc="Response time in milliseconds" ) # Error tracking error_message: Mapped[Optional[str]] = mapped_column( Text, doc="Error message if the request failed" ) # Timestamp timestamp: Mapped[datetime] = mapped_column( TIMESTAMP, nullable=False, server_default=func.now(), doc="When the request was made" ) # Indexes __table_args__ = ( Index("idx_api_audit_user", "user_id"), Index("idx_api_audit_endpoint", "endpoint"), Index("idx_api_audit_timestamp", "timestamp"), Index("idx_api_audit_status", "response_status"), ) def __repr__(self) -> str: """String representation of the audit log entry.""" return f""