Add audit schema and triggers for API tables

docker-compose.yml

@@ -42,7 +42,8 @@ services:
       # to new `SECRET` run openssl rand -hex 32
       SECRET: b8a3054ba3457614e95a88cc0807384430c1b338a54e95e4245f41e060da68bc
       ACCESS_TOKEN_EXPIRE_MINUTES: 30
-      TEST_ENVIRONMENT: 'False'
+      TEST_ENVIRONMENT: 'True'
+      DB_AUDIT_LOGS_ENABLED: 'True'
       MAIL_USERNAME: ''
       MAIL_FROM: admin@api-pgd.gov.br
       MAIL_PORT: 25

src/api.py

@@ -21,6 +21,8 @@
 from db_config import (
     check_db_connection,
     create_db_and_tables,
+    create_audit_ddl,
+    remove_audit_triggers,
     DbContextManager,
     get_db,
 )
@@ -34,7 +36,7 @@
     os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", DEFAULT_TOKEN_EXPIRE_MINS)
 )
 TEST_ENVIRONMENT = os.environ.get("TEST_ENVIRONMENT", "False") == "True"
-
+DB_AUDIT_LOGS_ENABLED = os.environ.get("DB_AUDIT_LOGS_ENABLED", "False") == "True"
 # ## INIT --------------------------------------------------
 
 with open(
@@ -66,6 +68,10 @@ async def lifespan(application: FastAPI):  # pylint: disable=unused-argument
     """Executa as rotinas de inicialização da API."""
     try:
         await create_db_and_tables()
+        if DB_AUDIT_LOGS_ENABLED:
+            await create_audit_ddl()
+        else:
+            await remove_audit_triggers()
         await crud_auth.init_user_admin()
     except OperationalError as exception:
         logger.error("A inicialização do banco de dados falhou: %s", exception)

src/db_audit.py

@@ -0,0 +1,79 @@
+AUDIT_DDL = """
+    CREATE SCHEMA IF NOT EXISTS auditoria;
+
+    CREATE TABLE IF NOT EXISTS auditoria.auditoria_db (
+        id SERIAL PRIMARY KEY,
+        operacao TEXT,
+        tabela TEXT,
+        registro_antigo JSONB,
+        registro_novo JSONB,
+        usuario TEXT,
+        data_operacao TIMESTAMPTZ DEFAULT now()
+    );
+
+    CREATE OR REPLACE FUNCTION auditoria.fn_auditoria() RETURNS TRIGGER AS $$
+    BEGIN
+        IF TG_OP = 'INSERT' THEN
+            INSERT INTO auditoria.auditoria_db (operacao, tabela, registro_novo, usuario)
+            VALUES ('INSERT', TG_TABLE_NAME, to_jsonb(NEW), current_user);
+
+        ELSIF TG_OP = 'UPDATE' THEN
+            INSERT INTO auditoria.auditoria_db (operacao, tabela, registro_antigo, registro_novo, usuario)
+            VALUES ('UPDATE', TG_TABLE_NAME, to_jsonb(OLD), to_jsonb(NEW), current_user);
+
+        ELSIF TG_OP = 'DELETE' THEN
+            INSERT INTO auditoria.auditoria_db (operacao, tabela, registro_antigo, usuario)
+            VALUES ('DELETE', TG_TABLE_NAME, to_jsonb(OLD), current_user);
+        END IF;
+
+        RETURN NULL;
+    END;
+    $$ LANGUAGE plpgsql;
+
+    DROP TRIGGER IF EXISTS tr_auditoria_pt ON plano_trabalho;
+    DROP TRIGGER IF EXISTS tr_auditoria_pe ON plano_entregas;
+    DROP TRIGGER IF EXISTS tr_auditoria_part ON participante;
+    DROP TRIGGER IF EXISTS tr_auditoria_us ON users;
+    DROP TRIGGER IF EXISTS tr_auditoria_co ON contribuicao;
+    DROP TRIGGER IF EXISTS tr_auditoria_en ON entrega;
+    DROP TRIGGER IF EXISTS tr_auditoria_are ON avaliacao_registros_execucao;
+
+
+    CREATE TRIGGER tr_auditoria_pt
+    AFTER INSERT OR UPDATE OR DELETE ON plano_trabalho
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_pe
+    AFTER INSERT OR UPDATE OR DELETE ON plano_entregas
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_part
+    AFTER INSERT OR UPDATE OR DELETE ON participante
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_us
+    AFTER INSERT OR UPDATE OR DELETE ON users
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_co
+    AFTER INSERT OR UPDATE OR DELETE ON contribuicao
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_en
+    AFTER INSERT OR UPDATE OR DELETE ON entrega
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+
+    CREATE TRIGGER tr_auditoria_are
+    AFTER INSERT OR UPDATE OR DELETE ON avaliacao_registros_execucao
+    FOR EACH ROW EXECUTE FUNCTION auditoria.fn_auditoria();
+"""
+
+REMOVE_AUDIT_TRIGGERS = """
+    DROP TRIGGER IF EXISTS tr_auditoria_pt ON plano_trabalho;
+    DROP TRIGGER IF EXISTS tr_auditoria_pe ON plano_entregas;
+    DROP TRIGGER IF EXISTS tr_auditoria_part ON participante;
+    DROP TRIGGER IF EXISTS tr_auditoria_us ON users;
+    DROP TRIGGER IF EXISTS tr_auditoria_co ON contribuicao;
+    DROP TRIGGER IF EXISTS tr_auditoria_en ON entrega;
+    DROP TRIGGER IF EXISTS tr_auditoria_are ON avaliacao_registros_execucao;
+"""

src/db_config.py

@@ -8,6 +8,7 @@
 from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.sql import text
+from db_audit import AUDIT_DDL, REMOVE_AUDIT_TRIGGERS
 
 SQLALCHEMY_DATABASE_URL = os.environ["SQLALCHEMY_DATABASE_URL"]
 
@@ -29,6 +30,16 @@ async def create_db_and_tables():
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
+async def create_audit_ddl():
+
+    async with engine.begin() as conn:
+        await conn.execute(text(AUDIT_DDL))
+
+async def remove_audit_triggers():
+
+    async with engine.begin() as conn:
+        await conn.execute(text(REMOVE_AUDIT_TRIGGERS))
+
 
 async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
     """Retorna a sessão do banco de dados.