Especificación Técnica: `audit_log`¶

Esquema: erp
Base de Datos: core
Servicio: svc-identidad
Tabla Legacy Origen: bd_erp.audit_log
Propósito: Bitácora de auditoría particionada por mes, ubicada en la base erp de cada producto (fuera de core).

1. Justificación y Mejoras de Arquitectura¶

Vive en el esquema erp de la base del producto (no en core); no comparte transaccionabilidad con el núcleo.
Particionada por rango de fecha (mes) con PK compuesta (id, fecha) para escritura mantenible.
Snapshot de nombres y códigos (modulo/menu/accion/actor) para auditar aunque el origen sea eliminado.
JSONB datos_antes/datos_despues/archivos para antes-después y adjuntos.

2. Definiciones de Implementación¶

DBMLSQLSQLAlchemyAlembic

Table erp.audit_log { id uuid [pk, default: gen_random_uuid()] empresa_id bigint [not null] fecha timestamptz [not null, default: now()] menu_id uuid [not null] accion_id uuid [not null] modulo_id uuid actor_user_id uuid [not null] actor_username varchar [not null] actor_nombre varchar [not null] actor_email varchar [not null] modulo_codigo varchar modulo_nombre varchar menu_codigo varchar menu_nombre varchar accion_codigo varchar accion_nombre varchar metodo_http varchar endpoint varchar referencia_tipo varchar referencia_id uuid detalle text datos_antes jsonb datos_despues jsonb archivos jsonb trace_id uuid created_at timestamptz [default: now()]

Indexes {
    (empresa_id) [name: 'ix_audit_log_empresa_id']
    (empresa_id, fecha) [name: 'ix_audit_log_empresa_fecha']
    (actor_user_id) [name: 'ix_audit_log_actor_user_id']
    (menu_id) [name: 'ix_audit_log_menu_id']
    (modulo_codigo) [name: 'ix_audit_log_modulo_codigo']
    (referencia_tipo, referencia_id) [name: 'ix_audit_log_referencia']
    (trace_id) [name: 'ix_audit_log_trace_id']
}

}

CREATE TABLE IF NOT EXISTS erp.audit_log (
    id                  UUID           NOT NULL DEFAULT gen_random_uuid(),
    empresa_id          BIGINT         NOT NULL,
    fecha               TIMESTAMPTZ    NOT NULL DEFAULT now(),
    menu_id             UUID           NOT NULL,
    accion_id           UUID           NOT NULL,
    modulo_id           UUID,
    actor_user_id       UUID           NOT NULL,
    actor_username      VARCHAR(50),
    actor_nombre        VARCHAR(200),
    actor_email         VARCHAR(150),
    modulo_codigo       VARCHAR(50),
    modulo_nombre       VARCHAR(150),
    menu_codigo         VARCHAR(50),
    menu_nombre         VARCHAR(150),
    accion_codigo       VARCHAR(50),
    accion_nombre       VARCHAR(100),
    metodo_http         VARCHAR(10),
    endpoint            VARCHAR(300),
    referencia_tipo     VARCHAR(50),
    referencia_id       UUID,
    detalle             TEXT,
    datos_antes         JSONB,
    datos_despues       JSONB,
    archivos            JSONB,
    trace_id            UUID,
    created_at          TIMESTAMPTZ    DEFAULT now(),
    PRIMARY KEY (id, fecha)
) PARTITION BY RANGE (fecha);

-- Índices
CREATE INDEX IF NOT EXISTS ix_audit_log_empresa_id ON erp.audit_log (empresa_id);
CREATE INDEX IF NOT EXISTS ix_audit_log_empresa_fecha ON erp.audit_log (empresa_id, fecha);
CREATE INDEX IF NOT EXISTS ix_audit_log_actor_user_id ON erp.audit_log (actor_user_id);
CREATE INDEX IF NOT EXISTS ix_audit_log_menu_id ON erp.audit_log (menu_id);
CREATE INDEX IF NOT EXISTS ix_audit_log_modulo_codigo ON erp.audit_log (modulo_codigo);
CREATE INDEX IF NOT EXISTS ix_audit_log_referencia ON erp.audit_log (referencia_tipo, referencia_id);
CREATE INDEX IF NOT EXISTS ix_audit_log_trace_id ON erp.audit_log (trace_id);

-- Las particiones mensuales (audit_log_YYYYMM) se crean operativamente
-- previo al rango de fechas a registrar.

-- Comentarios
COMMENT ON TABLE erp.audit_log IS 'Bitácora de auditoría particionada por mes (fecha) del dominio identidad.';

from datetime import datetime
from typing import Optional
import uuid

from sqlalchemy import (
    BigInteger,
    CheckConstraint,
    DateTime,
    Index,
    String,
    Text,
    UniqueConstraint,
    text,
)
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class AuditLogERP(Base):
    __tablename__ = "audit_log"
    __table_args__ = (
        Index("ix_audit_log_empresa_id", "empresa_id"),
        Index("ix_audit_log_empresa_fecha", "empresa_id", "fecha"),
        Index("ix_audit_log_actor_user_id", "actor_user_id"),
        Index("ix_audit_log_menu_id", "menu_id"),
        Index("ix_audit_log_modulo_codigo", "modulo_codigo"),
        Index("ix_audit_log_referencia", "referencia_tipo", "referencia_id"),
        Index("ix_audit_log_trace_id", "trace_id"),
        {"schema": "erp"},
    )

    id: Mapped[uuid.UUID] = mapped_column(
        postgresql.UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )
    empresa_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    fecha: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=text("now()")
    )
    menu_id: Mapped[uuid.UUID] = mapped_column(postgresql.UUID(as_uuid=True), nullable=False)
    accion_id: Mapped[uuid.UUID] = mapped_column(postgresql.UUID(as_uuid=True), nullable=False)
    modulo_id: Mapped[Optional[uuid.UUID]] = mapped_column(postgresql.UUID(as_uuid=True))
    actor_user_id: Mapped[uuid.UUID] = mapped_column(postgresql.UUID(as_uuid=True), nullable=False)
    actor_username: Mapped[Optional[str]] = mapped_column(String(50))
    actor_nombre: Mapped[Optional[str]] = mapped_column(String(200))
    actor_email: Mapped[Optional[str]] = mapped_column(String(150))
    modulo_codigo: Mapped[Optional[str]] = mapped_column(String(50))
    modulo_nombre: Mapped[Optional[str]] = mapped_column(String(150))
    menu_codigo: Mapped[Optional[str]] = mapped_column(String(50))
    menu_nombre: Mapped[Optional[str]] = mapped_column(String(150))
    accion_codigo: Mapped[Optional[str]] = mapped_column(String(50))
    accion_nombre: Mapped[Optional[str]] = mapped_column(String(100))
    metodo_http: Mapped[Optional[str]] = mapped_column(String(10))
    endpoint: Mapped[Optional[str]] = mapped_column(String(300))
    referencia_tipo: Mapped[Optional[str]] = mapped_column(String(50))
    referencia_id: Mapped[Optional[uuid.UUID]] = mapped_column(postgresql.UUID(as_uuid=True))
    detalle: Mapped[Optional[str]] = mapped_column(Text)
    datos_antes: Mapped[Optional[object]] = mapped_column(postgresql.JSONB)
    datos_despues: Mapped[Optional[object]] = mapped_column(postgresql.JSONB)
    archivos: Mapped[Optional[object]] = mapped_column(postgresql.JSONB)
    trace_id: Mapped[Optional[uuid.UUID]] = mapped_column(postgresql.UUID(as_uuid=True))
    created_at: Mapped[Optional[datetime]] = mapped_column(
        DateTime(timezone=True), server_default=text("now()")
    )

"""revision: core_id_0023
create table erp.audit_log (particionado por rango de fecha)

"""

from alembic import op
import sqlalchemy as sa


revision: str = "core_id_0023"
down_revision: str | None = "core_id_0022"
branch_labels: str | None = None
depends_on: str | None = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE IF NOT EXISTS erp.audit_log (
            id                  UUID           NOT NULL DEFAULT gen_random_uuid(),
            empresa_id          BIGINT         NOT NULL,
            fecha               TIMESTAMPTZ    NOT NULL DEFAULT now(),
            menu_id             UUID           NOT NULL,
            accion_id           UUID           NOT NULL,
            modulo_id           UUID,
            actor_user_id       UUID           NOT NULL,
            actor_username      VARCHAR(50),
            actor_nombre        VARCHAR(200),
            actor_email         VARCHAR(150),
            modulo_codigo       VARCHAR(50),
            modulo_nombre       VARCHAR(150),
            menu_codigo         VARCHAR(50),
            menu_nombre         VARCHAR(150),
            accion_codigo       VARCHAR(50),
            accion_nombre       VARCHAR(100),
            metodo_http         VARCHAR(10),
            endpoint            VARCHAR(300),
            referencia_tipo     VARCHAR(50),
            referencia_id       UUID,
            detalle             TEXT,
            datos_antes         JSONB,
            datos_despues       JSONB,
            archivos            JSONB,
            trace_id            UUID,
            created_at          TIMESTAMPTZ    DEFAULT now(),
            PRIMARY KEY (id, fecha)
        ) PARTITION BY RANGE (fecha);
        """
    )
    op.create_index("ix_audit_log_empresa_id", "audit_log", ["empresa_id"], schema="erp")
    op.create_index("ix_audit_log_empresa_fecha", "audit_log", ["empresa_id", "fecha"], schema="erp")
    op.create_index("ix_audit_log_actor_user_id", "audit_log", ["actor_user_id"], schema="erp")
    op.create_index("ix_audit_log_menu_id", "audit_log", ["menu_id"], schema="erp")
    op.create_index("ix_audit_log_modulo_codigo", "audit_log", ["modulo_codigo"], schema="erp")
    op.create_index("ix_audit_log_referencia", "audit_log", ["referencia_tipo", "referencia_id"], schema="erp")
    op.create_index("ix_audit_log_trace_id", "audit_log", ["trace_id"], schema="erp")


def downgrade() -> None:
    op.drop_table("audit_log", schema="erp")

Especificación Técnica: audit_log¶

1. Justificación y Mejoras de Arquitectura¶

2. Definiciones de Implementación¶

Especificación Técnica: `audit_log`¶