audit-recorder 0.2.0

Generic audit library for applications

🔐 audit-recorder

Une librairie complète et flexible pour auditer les actions dans vos applications Python, avec support complet async/sync et extensibilité.

✨ Features

• 📋 Audit complet: CREATE, UPDATE, DELETE, et actions custom
• 🔄 Async/Sync: Support natif de async et sync
• 🎯 Décorateurs simples: @audit, @transactional, @audit_entity
• ⚙️ Configuration centralisée: AuditConfig pour simplifier le setup
• 🧩 Extensible: Registry de sérializers pour vos types custom
• Flexible: Support multi-formats de tokens et contextes utilisateur
• 💾 Champs dynamiques: extra_fields pour enrichir vos logs

🚀 Installation

pip install audit-recorder

---

📖 Quick Start

1️⃣ Configuration simple

from audit_recorder import AuditConfig
from sqlalchemy import create_engine

# Initialiser la config
config = AuditConfig(
    user_id_extractor=lambda token: token.sub,
    user_email_extractor=lambda token: token.email,
)
config.init()

2️⃣ Enregistrer vos entités

from audit_recorder import audit_entity
from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import declarative_base

Base = declarative_base()

@audit_entity
class Article(Base):
    __tablename__ = "articles"
    id = Column(Integer, primary_key=True)
    title = Column(String)

3️⃣ Décorer vos services

from audit_recorder import audit, transactional

@transactional  # Commit/rollback auto
@audit(action='UPDATE', resource_type='Article')
def update_article(session, article_id: int, title: str, id_token):
    article = session.query(Article).get(article_id)
    article.title = title
    # ✅ Pas besoin de commit, @transactional s'en charge
    return article

Paramètres obligatoires dans la signature de la fonction décorée :

• session : doit être nommé session (instance Session SQLAlchemy)
• id_token : doit correspondre au user_param configuré dans AuditConfig (défaut : 'id_token')

4️⃣ Consulter les logs

from audit_recorder import service

# Récupérer les logs d'un article
logs = service.query_logs(
    db,
    resource_type='Article',
    resource_id='42',
    page=1,
    limit=20
)

for log in logs.results:
    print(f"{log.action} on {log.resource_type} by {log.user_email}")
    print(f"  Old: {log.old_values}")
    print(f"  New: {log.new_values}")
    print(f"  Changes: {log.changes}")

---

🔧 Configuration centralisée (AuditConfig)

AuditConfig centralise tous les paramétrages de la librairie dans une seule classe, éliminant le boilerplate.

Attributs

Paramètre	Type	Description
user_id_extractor	Callable	Fonction pour extraire l'ID utilisateur
user_email_extractor	Callable	Fonction pour extraire l'email utilisateur
extra_fields_resolver	Callable	Fonction pour extraire les champs dynamiques
default_skip_actions	list	Actions à ignorer par défaut
custom_serializers	dict	Sérializers pour vos types custom

Exemples

Avec champs dynamiques:

def extract_audit_fields(action, resource_type, resource_id, result):
    return {
        "ip_address": request.remote_addr,
        "user_agent": request.headers.get("user-agent"),
        "timestamp": datetime.now().isoformat(),
    }

config = AuditConfig(
    user_id_extractor=lambda token: token.sub,
    user_email_extractor=lambda token: token.email,
    extra_fields_resolver=extract_audit_fields,
)
config.init()

Avec sérializers custom:

from decimal import Decimal
from datetime import date

config = AuditConfig(
    user_id_extractor=lambda token: token.sub,
    custom_serializers={
        date: lambda d: d.isoformat(),
        Decimal: lambda dec: float(dec),
    }
)
config.init()

---

🏷️ Enregistrement automatique d'entités (@audit_entity)

Le décorateur @audit_entity enregistre automatiquement le resolver pour une entité, sans code répétitif.

Simple (auto-détection)

Sans loader explicite, @audit_entity cherche dans l'ordre :

1. Une classmethod load(cls, session, resource_id) sur la classe
2. Sinon, fallback sur session.query(cls).get(resource_id)

@audit_entity
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # Pas de load() → utilise session.query(User).get(resource_id)

Avec load() défini :

@audit_entity
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)

    @classmethod
    def load(cls, session, user_id):
        # Chargé automatiquement pour les snapshots avant/après
        return session.query(cls).options(joinedload(cls.role)).get(user_id)

Avec loader personnalisé

def load_user(session, user_id):
    return session.query(User).filter(User.uuid == user_id).first()

@audit_entity(loader=load_user)
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    uuid = Column(String, unique=True)

Avec resource_type custom

@audit_entity(resource_type="CustomUser")
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)

Avec un identifiant autre que id

Par défaut, @audit_entity utilise l'attribut id comme identifiant de la ressource. C'est cet attribut qui est utilisé :

• pour extraire l'ID depuis l'objet retourné par une action CREATE (quand id_param n'est pas renseigné dans @audit)
• pour charger les snapshots avant/après via le loader

Si ta clé primaire ne s'appelle pas id, précise-le avec identifier :

@audit_entity(identifier='uuid')
class Article(Base):
    __tablename__ = "articles"
    uuid = Column(String, primary_key=True)
    title = Column(String)

Ou délègue entièrement la résolution au loader :

@audit_entity(loader=lambda session, rid: session.query(Article).filter_by(uuid=rid).first())
class Article(Base):
    __tablename__ = "articles"
    uuid = Column(String, primary_key=True)

---

🔄 Transaction automatique (@transactional)

Le décorateur @transactional gère automatiquement le commit et le rollback, éliminant les bugs "changement sans audit" ou "audit orphelin".

@transactional
@audit(action='UPDATE', resource_type='Article')
def update_article(session, article_id: int, title: str, id_token):
    article = session.query(Article).get(article_id)
    article.title = title
    # ✅ Commit auto, rollback auto en cas d'erreur
    # ❌ Pas de session.commit() nécessaire
    return article

# Utilisation
try:
    article = update_article(db_session, 1, "New Title", token)
    # Déjà committée !
except ValueError:
    # Déjà rollbackée !
    pass

Support Async/Sync automatique

# Async - il suffit de passer async
@transactional
@audit(action='UPDATE', resource_type='Article')
async def update_article(session, article_id: int, id_token):
    # Marche aussi en async !
    pass

# Sync - marche aussi normalement
@transactional
@audit(action='UPDATE', resource_type='Article')
def update_article(session, article_id: int, id_token):
    # Marche en sync
    pass

---

🎨 Sérializers extensibles

Enregistrez des sérializers custom pour vos types propriétaires.

Fonction simple

from audit_recorder import register_serializer
from datetime import date

# Enregistrer pour un type simple
register_serializer(date, lambda d: d.isoformat())

# Ensuite, les dates sont auto-sérialisées :
article.published_on = date(2024, 1, 15)
# Audit stocke : {"published_on": "2024-01-15"} ✅

Classe custom

from audit_recorder import Serializer, register_serializer

class DateRangeSerializer(Serializer):
    def serialize(self, value):
        return {
            'start': value.start.isoformat(),
            'end': value.end.isoformat(),
        }

register_serializer(DateRange, DateRangeSerializer())

Récupérer un sérializer

from audit_recorder import get_serializer

serializer = get_serializer(date)
if serializer:
    result = serializer.serialize(date(2024, 1, 15))
    # result: "2024-01-15"

---

🗂️ Champs dynamiques (extra_fields)

Enrichissez vos logs avec des données contextuelles dynamiques.

Définir un resolver

from audit_recorder import AuditConfig
from datetime import datetime

def extract_audit_fields(action, resource_type, resource_id, result):
    return {
        "timestamp": datetime.now().isoformat(),
        "user_role": current_user.role,
        "ip_address": request.remote_addr,
    }

config = AuditConfig(
    extra_fields_resolver=extract_audit_fields,
)
config.init()

Accéder aux champs

logs = service.query_logs(db, resource_type="Article")

for log in logs.results:
    # Accès direct (via model_validator flatten_extra_fields)
    print(f"IP: {log.ip_address}")
    print(f"Role: {log.user_role}")
    print(f"Timestamp: {log.timestamp}")
    
    # Ou via le dict
    print(f"All extra: {log.extra_fields}")

---

⏭️ Skip audit conditionnellement

Désactiver l'audit dans un contexte spécifique.

Action unique

from audit_recorder import skip_audit

with skip_audit('EXPORT'):
    export_data(session)  # ← Pas auditée
    
# Les autres actions sont auditées normalement
update_article(session)  # ← Auditée

Plusieurs actions

with skip_audit('EXPORT', 'SYNC', 'CLEANUP'):
    do_batch_operation(session)  # ← Aucune de ces 3 actions auditée

Tout désactiver

with skip_audit():  # Pas d'argument = tout désactiver
    do_sensitive_operation(session)  # ← Aucune action auditée

Imbriqué (fusion)

with skip_audit('CREATE'):
    create_user(session)  # ← CREATE non auditée
    
    with skip_audit('UPDATE'):
        update_user(session)  # ← Ni CREATE ni UPDATE auditées
    
    update_user(session)  # ← CREATE non auditée, UPDATE auditée

delete_user(session)  # ← DELETE auditée

---

🗄️ Requêtes sur les logs

Query simple

from audit_recorder import service

logs = service.query_logs(
    db,
    resource_type='Article',
    resource_id='42',
    action='UPDATE',
    user_id='user_123',
    date_from=datetime(2024, 1, 1),
    date_to=datetime(2024, 12, 31),
    page=1,
    limit=50,
)

print(f"Total: {logs.total}")
print(f"Pages: {logs.pages}")
for log in logs.results:
    print(log)

Query par ID

log = service.query_log_by_id(db, 'audit_id_123')
print(log)

Enrichir les logs

Les logs stockent des IDs bruts (resource_id, user_id). Pour afficher des labels lisibles sur une interface, passez un populate : une fonction appelée sur chaque DTO après chargement, qui peut y ajouter des champs arbitraires.

def populate(log: AuditLogDTO, db) -> AuditLogDTO:
    if log.resource_type == 'Article':
        article = db.get(Article, log.resource_id)
        log.resource_label = article.title if article else log.resource_id
    if log.user_id:
        user = db.get(User, log.user_id)
        log.user_display_name = user.full_name if user else log.user_email
    return log

logs = service.query_logs(db, resource_type='Article', populate=populate)

for log in logs.results:
    print(log.resource_label)    # "Mon article" au lieu de "42"
    print(log.user_display_name) # "Jean Dupont" au lieu de "user_123"

Fonctionne aussi sur query_log_by_id :

log = service.query_log_by_id(db, 'audit_id_123', populate=populate)
print(log.resource_label)

Les champs ajoutés par populate sont directement accessibles sur le DTO (log.resource_label) grâce au extra='allow' de Pydantic.

Structure d'un log

print(log.id)              # str UUID
print(log.action)          # str "CREATE", "UPDATE", "DELETE", etc.
print(log.resource_type)   # str "Article", "User", etc.
print(log.resource_id)     # str | None
print(log.user_id)         # str | None
print(log.user_email)      # str | None
print(log.old_values)      # dict | None Snapshot avant
print(log.new_values)      # dict | None Snapshot après
print(log.changes)         # dict | None Diff précis
print(log.extra_fields)    # dict | None Champs dynamiques
print(log.created_at)      # datetime

---

📦 Schema & Migrations

Sans Alembic (dev / nouveau projet)

Crée la table directement via l'engine SQLAlchemy :

from sqlalchemy import create_engine
from audit_recorder.models import Base, AuditLog

engine = create_engine('postgresql://...')
Base.metadata.create_all(engine, tables=[AuditLog.__table__])

Avec Alembic (production)

Expose le Base d'audit_recorder dans env.py pour que l'autogenerate détecte la table audit_logs :

# alembic/env.py
from audit_recorder.models import Base as AuditBase
from myapp.models import Base as AppBase

target_metadata = [AppBase.metadata, AuditBase.metadata]

Ensuite :

alembic revision --autogenerate -m "add audit_logs table"
alembic upgrade head

Schema

CREATE TABLE audit_logs (
  -- Identification
  id VARCHAR(36) PRIMARY KEY,
  
  -- QUI (utilisateur)
  user_id VARCHAR(255),
  user_email VARCHAR(255),
  
  -- QUOI (action)
  action VARCHAR(50) NOT NULL,
  resource_type VARCHAR(100) NOT NULL,
  resource_id VARCHAR(255),
  
  -- SNAPSHOTS (changements)
  old_values JSON,
  new_values JSON,
  changes JSON,
  
  -- QUAND
  created_at DATETIME NOT NULL,
  
  -- EXTRA
  extra_fields JSON
);

---

🧪 Testing

import pytest
from audit_recorder import skip_audit, AuditResult

def test_audit_creation():
    @audit(action='CREATE', resource_type='TestModel')
    def create_model(session, id_token):
        return {'id': 1}
    
    # Tester normalement, l'audit se fera
    result = create_model(session=mock_session, id_token=mock_token)
    assert result['id'] == 1

def test_skip_audit():
    @audit(action='UPDATE', resource_type='TestModel')
    def update_model(session, id_token):
        return {'id': 1}
    
    # Désactiver l'audit dans un test
    with skip_audit('UPDATE'):
        result = update_model(session=mock_session, id_token=mock_token)
        # L'audit n'a pas été enregistré

def test_audit_result_skip():
    @audit(action='EXPORT', resource_type='Data')
    def export_data(session, id_token):
        return AuditResult.skip(data=[1, 2, 3], reason="Export is non-auditable")
    
    result = export_data(session=mock_session, id_token=mock_token)
    # L'audit n'a pas été enregistré

---

🧹 Linting & Formatting

# Format avec ruff
ruff check . --fix
ruff format .

# Tester
pytest

# Builder la librairie
python -m pip install --upgrade build
python -m build

---

📚 Documentation complète

Modules

• audit_decorator.py - Décorateur @audit
• config.py - Configuration centralisée
• entity.py - Décorateur @audit_entity
• transactional_decorator.py - Décorateur @transactional
• serializers.py - Registry de sérializers
• context.py - Context vars et controls
• models.py - Modèles SQLAlchemy et Pydantic
• service.py - Requêtes et logs
• resolver.py - Résolution d'entités

---

📄 Licence

MIT License - voir LICENSE pour les détails.

---