AuditLogProtocol

Protocol defining the contract for audit log implementations.

This is the primary port for audit logging in the hexagonal architecture.
Implementations handle different storage backends.

Implementations include:
- DatabaseAuditAdapter: Writes to ActivityLog table (Indie mode)
- FileAuditAdapter: Writes to JSONL file for Splunk/Filebeat
- ElasticAuditAdapter: Writes directly to Elasticsearch

Example usage:
    audit: AuditLogProtocol = container.get(AuditLogProtocol)

    # Log an action
    entry_id = await audit.log(
        user_id="user-001",
        action="create",
        doctype="Invoice",
        document_id="INV-001",
    )

    # Query recent actions
    entries = await audit.query(
        filters={"user_id": "user-001", "action": "update"},
        limit=50,
    )

Source: audit.py

Methods

log

async def log(self,
    user_id: str,
    action: str,
    doctype: str,
    document_id: str,
    changes: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None,
) -> str

Log an audit entry.

    Creates an immutable record of the action. This should never fail
    silently - audit logging failures should be treated seriously.

    Args:
        user_id: ID of the user performing the action
        action: Type of action ("create", "read", "update", "delete")
        doctype: Name of the DocType
        document_id: ID of the document
        changes: Optional field changes for updates
        metadata: Optional context (request_id, ip, user_agent, etc.)

    Returns:
        ID of the created audit entry

query

async def query(self,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[AuditEntry]

Query audit entries with filters.

    Retrieves audit entries matching the given criteria.
    Results are ordered by timestamp descending (newest first).

    Args:
        filters: Optional filter criteria:
            - user_id: Filter by user
            - action: Filter by action type
            - doctype: Filter by DocType
            - document_id: Filter by specific document
            - from_timestamp: Filter entries after this time
            - to_timestamp: Filter entries before this time
        limit: Maximum entries to return (default 50, max 1000)
        offset: Number of entries to skip for pagination

    Returns:
        List of matching AuditEntry objects