SourceShareOSV1/server/audit.py

import json
from typing import Any, Optional

from flask import request

from .core import isoformat, utcnow
from .db import execute


def audit(
    actor_type: str,
    actor_id: int,
    action: str,
    resource_type: str,
    resource_id: str,
    before: Optional[dict[str, Any]],
    after: Optional[dict[str, Any]],
) -> None:
    execute(
        """
        INSERT INTO audit_logs (actor_type, actor_id, action, resource_type, resource_id, before_json, after_json, ip, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            actor_type,
            actor_id,
            action,
            resource_type,
            resource_id,
            json.dumps(before, ensure_ascii=False) if before is not None else None,
            json.dumps(after, ensure_ascii=False) if after is not None else None,
            request.headers.get("X-Forwarded-For", request.remote_addr),
            isoformat(utcnow()),
        ),
    )