audit-recorder 0.5.1

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é.

---

Objectif

audit-recorder trace automatiquement les actions sur vos entités SQLAlchemy, sans modifier votre code métier. Un décorateur @audit suffit pour capturer le snapshot avant/après chaque appel, calculer le diff et enregistrer le log avec le contexte utilisateur.

✨ Features

• 📋 Audit complet — CREATE, UPDATE, DELETE, et actions custom
• 🔄 Async/Sync — support natif, détection automatique
• 🎯 Décorateurs simples — @audit, @transactional, @audit_entity
• ⚙️ Configuration centralisée — AuditConfig pour tout paramétrer en un endroit
• 🔑 Flexible — support multi-formats de tokens et contextes utilisateur
• 💾 Champs dynamiques — extra_fields pour enrichir vos logs sans migration
• 🏷️ Identifiants simples — resource_id scalaire (str ou int)
• 🗄️ Multi-base — PostgreSQL (JSONB), MariaDB/MySQL, SQLite

→ Quickstart — exemple complet en 5 minutes

---

Données enregistrées par défaut

Chaque log contient les champs suivants, renseignés automatiquement :

log.id            # str — UUID
log.action        # str — "CREATE", "UPDATE"…
log.resource_type # str — "Article", "User"…
log.resource_id   # str | None — identifiant de la ressource
log.user_id       # str | None — extrait du token via user_id_extractor
log.user_email    # str | None — extrait du token via user_email_extractor
log.old_values    # dict | None — snapshot avant l'appel
log.new_values    # dict | None — snapshot après l'appel
log.changes       # dict | None — diff champ par champ
log.extra_fields  # dict | None — champs contextuels (via extra_fields_resolver)
log.created_at    # datetime — horodatage UTC

old_values est None sur les CREATE, new_values est None sur les DELETE. Passez track_data=False au décorateur pour ne logger que l'action sans capturer les données.

---

Installation

pip install audit-recorder

---

Configuration

1. Créer la table

Sans Alembic (dev / nouveau projet)

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

engine = create_engine('postgresql://user:pass@localhost/mydb')
Base.metadata.create_all(engine, tables=[AuditLog.__table__])

Avec Alembic (production)

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

target_metadata = [AppBase.metadata, AuditBase.metadata]

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

2. Initialiser AuditConfig

from audit_recorder import AuditConfig

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

Paramètre	Type	Description
user_param	str	Nom du paramètre portant le token (défaut : 'id_token')
user_id_extractor	Callable	Extrait l'ID utilisateur depuis le token
user_email_extractor	Callable	Extrait l'email utilisateur depuis le token
extra_fields_resolver	Callable	Ajoute des champs contextuels à chaque log
model	type	Sous-classe de AuditLog avec des colonnes métier indexées

3. Compatibilité bases de données

Base de données	Support	Type JSON utilisé
PostgreSQL	✅	JSONB (binaire, indexable)
MariaDB / MySQL	✅	JSON
SQLite	✅	JSON

Le type est sélectionné automatiquement selon le dialecte de l'URL de connexion.

---

Enregistrer vos entités à auditer

@audit_entity enregistre le resolver qui charge les snapshots avant/après pour une entité.

Paramètre	Description
snapshot	Options SQLAlchemy pour le snapshot (liste ou lambda cls: [...])
loader	Fonction de chargement custom (session, resource_id) -> entity
identifier	Nom de la PK si différent de id
resource_type	Nom de la ressource si différent du nom de la classe

Cas minimal — sans configuration, @audit_entity cherche une classmethod load() sur la classe, puis fallback sur session.query(cls).get(resource_id) :

from audit_recorder import audit_entity

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

Avec une classmethod load() :

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

    @classmethod
    def load(cls, session, resource_id):
        return session.query(cls).options(joinedload(cls.role)).get(resource_id)

Avec snapshot de relations — snapshot= ouvre une session fraîche (évite l'identity map) et applique les options SQLAlchemy. noload('*') est ajouté automatiquement pour exclure les relations non listées :

from sqlalchemy.orm import selectinload

@audit_entity(snapshot=lambda cls: [
    selectinload(cls.category).load_only(Category.name),
    selectinload(cls.tags),
])
class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)

Le lambda cls: [...] est recommandé pour éviter les forward references : les attributs de la classe ne sont résolus qu'au premier appel, après que tous les mappers SQLAlchemy sont initialisés.

snapshot= est ignoré si loader= est fourni explicitement.

⚠️ Données sensibles : le loader sérialise toutes les colonnes chargées. Utilisez load_only() pour exclure les champs sensibles (mots de passe, tokens…) des snapshots.

Avec un resource_type custom :

@audit_entity(resource_type='MY_RESOURCE')
class MyModel(Base): ...

⚠️ Sans resource_type explicite, la clé est le cls.__name__ en majuscules, sans underscore. DemandeAcces → "DEMANDEACCES". Si le nom contient un underscore, passez resource_type explicitement aux deux décorateurs.

Avec une PK différente de id :

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

resource_id est scalaire (str ou int). Les clés primaires composites ne sont pas prises en charge.

---

Décorer vos services

@audit capture les snapshots avant/après chaque appel et enregistre le log.

from audit_recorder import audit

@audit(action='UPDATE', resource_type='Article', id_param='article_id')
def update_article(session, article_id: int, title: str, id_token):
    article = session.query(Article).get(article_id)
    article.title = title
    session.flush()
    return article

Paramètre	Description
action	Nom de l'action ('CREATE', 'UPDATE', 'DELETE', custom…)
resource_type	Nom de la ressource — doit correspondre à @audit_entity
id_param	Chemin vers l'ID avant l'appel : 'article_id', 'payload.id', ou callable
track_data	True (défaut) : capture old/new/changes. False : log sans snapshot

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

• session : instance Session SQLAlchemy
• id_token : paramètre portant le token (configuré via user_param)

id_param — capture des snapshots avant/après

id_param indique comment récupérer l'ID avant l'appel pour charger old_values.

Cas	id_param	old_values	new_values
CREATE — retourne l'entité	omis	—	✅ résolu depuis le retour
CREATE — retourne un DTO	'payload.id'	—	✅
UPDATE / DELETE / custom	'article_id'	✅	✅

# Chemin simple
@audit(action='DELETE', resource_type='Article', id_param='article_id')
def delete_article(session, article_id: int, id_token): ...

# Chemin imbriqué (argument objet)
@audit(action='UPDATE', resource_type='Article', id_param='payload.article_id')
def update_article(session, payload: UpdatePayload, id_token): ...

# Callable pour une logique custom
@audit(action='UPDATE', resource_type='Article', id_param=lambda args: args['payload'].id)
def update_article(session, payload: UpdatePayload, id_token): ...

Note : avec snapshot= (session fraîche), un flush ne suffit pas toujours pour relire les changements — la visibilité dépend du commit.

track_data=False — log sans capture de données

@audit(action='READ', resource_type='Document', id_param='document_id', track_data=False)
def get_document(session, document_id: int, id_token):
    return session.get(Document, document_id)

Le log contient action, resource_type, resource_id, user_id, user_email et created_at, mais old_values, new_values et changes sont None.

---

Transaction automatique (@transactional)

@transactional gère le commit et le rollback automatiquement.

from audit_recorder import audit, transactional

@transactional
@audit(action='UPDATE', resource_type='Article', id_param='article_id')
def update_article(session, article_id: int, title: str, id_token):
    article = session.query(Article).get(article_id)
    article.title = title
    return article
    # ✅ commit auto si succès, rollback auto si exception

Fonctionne aussi en async :

@transactional
@audit(action='UPDATE', resource_type='Article')
async def update_article(session, article_id: int, id_token): ...

---

Étendre les données enregistrées

Champs dynamiques (extra_fields)

Enrichissez chaque log avec des données contextuelles via extra_fields_resolver. Aucune migration nécessaire — les données sont stockées dans la colonne JSON extra_fields et accessibles directement sur le DTO.

def extra_fields(action, resource_type, resource_id, result, user_context):
    return {
        'ip_address': request.remote_addr,
        'user_role': user_context.role if user_context else None,
    }

AuditConfig(
    user_id_extractor=lambda token: token.sub,
    extra_fields_resolver=extra_fields,
).init()

for log in logs.results:
    print(log.ip_address)   # accès direct sur le DTO
    print(log.extra_fields) # {"ip_address": "...", "user_role": "..."}

Colonnes dédiées pour les champs fréquemment filtrés

Pour indexer un champ de extra_fields, déclarez-le comme vraie colonne via une sous-classe de AuditLog. La lib le détecte automatiquement et l'y stocke directement à la place de extra_fields.

1. Déclarer la sous-classe

from audit_recorder import AuditLog
from sqlalchemy import Column, String

class MyAuditLog(AuditLog):
    __tablename__ = 'audit_logs'
    __table_args__ = {'extend_existing': True}
    tenant_id = Column(String(50), index=True)

2. Configurer

AuditConfig(
    model=MyAuditLog,
    extra_fields_resolver=lambda action, rt, rid, result, ctx, old, new: {
        'tenant_id': ctx.tenant_id if ctx else None,  # → colonne dédiée
        'ip_address': request.remote_addr,            # → extra_fields
    },
).init()

3. Migrer (Alembic, à la charge de l'utilisateur)

op.add_column('audit_logs', sa.Column('tenant_id', sa.String(50), nullable=True))
op.create_index('idx_audit_tenant', 'audit_logs', ['tenant_id'])

4. Filtrer — la lib route automatiquement selon que la clé est une colonne dédiée ou non :

# tenant_id → colonne SQL indexée | ip_address → JSON scan
service.query_logs(db, extra_fields_filter={'tenant_id': 'acme', 'ip_address': '1.2.3'})

Les champs promus en colonne dédiée sont absents de extra_fields. Sur MariaDB/SQLite à fort volume, les filtres JSON sur les champs non promus font un scan complet.

---

Consulter les logs

from audit_recorder import service

logs = service.query_logs(
    db,
    resource_type='Article',
    resource_id='42',
    action='UPDATE',
    page=1,
    limit=20,
)

print(f'Total : {logs.total} — Pages : {logs.pages}')
for log in logs.results:
    print(log.action, log.resource_id, log.user_email)
    print('  avant :', log.old_values)
    print('  après :', log.new_values)
    print('  diff  :', log.changes)

Paramètre	Type	Description
resource_type	str | list[str]	Type(s) de ressource
resource_id	str	Identifiant exact
action	str | list[str]	Action(s) à filtrer
user_id	str	ID utilisateur (insensible à la casse)
user_email	str	Email (recherche partielle, insensible à la casse)
date_from	datetime	Borne inférieure sur created_at
date_to	datetime	Borne supérieure sur created_at
page	int	Numéro de page (défaut : 1)
limit	int	Résultats par page (défaut : 20, max : 1000)
populate	Callable	Enrichissement des DTOs après chargement

Récupérer un log par ID :

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

Enrichir les résultats avec populate — appelé sur chaque DTO, utile pour ajouter des labels lisibles à partir des IDs bruts :

from audit_recorder.models import AuditLogDTO

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 str(log.resource_id)
    return log

logs = service.query_logs(db, resource_type='Article', populate=populate)
# Fonctionne aussi sur query_log_by_id
log = service.query_log_by_id(db, 'audit_id_123', populate=populate)

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

---

Filtres de précision optionnels

extra_fields_filter et values_filter filtrent directement en SQL sur les colonnes JSON. La pagination reste entièrement côté base.

# Recherche partielle sur extra_fields (mode ilike par défaut)
logs = service.query_logs(db, extra_fields_filter={'user_role': 'adm'})

# Égalité stricte sur new_values
logs = service.query_logs(
    db,
    values_filter={'value': 'published', 'fields': ['status']},
    values_filter_mode='exact',
)

# Notation pointée pour les champs imbriqués
logs = service.query_logs(
    db,
    values_filter={'value': '[e-mail]', 'fields': ['author.email']},
)

# Modes différents par filtre
logs = service.query_logs(
    db,
    extra_fields_filter={'user_role': 'adm'},
    extra_fields_filter_mode='ilike',
    values_filter={'value': 'published', 'fields': ['status']},
    values_filter_mode='exact',
)

Mode	Comportement
'ilike' (défaut)	Recherche partielle, insensible à la casse
'exact'	Égalité stricte, insensible à la casse

Compatibilité SQL par dialecte :

Dialecte	SQL généré pour column['key']
PostgreSQL	column ->> 'key'
MariaDB / MySQL	JSON_UNQUOTE(JSON_EXTRACT(column, '$.key'))
SQLite	JSON_EXTRACT(column, '$.key')

⚠️ JSON_UNQUOTE sur MySQL/MariaDB requiert SQLAlchemy >= 1.4.

---

Avancé

Désactiver l'audit (skip_audit)

from audit_recorder import skip_audit

with skip_audit('EXPORT'):              # une action
    export_data(session)

with skip_audit('EXPORT', 'SYNC'):      # plusieurs actions
    do_batch_operation(session)

with skip_audit():                      # tout désactiver
    do_sensitive_operation(session)

# Imbriqué — les actions désactivées s'accumulent
with skip_audit('CREATE'):
    with skip_audit('UPDATE'):
        ...  # CREATE et UPDATE désactivées

Cohérence des types : skip_audit compare par égalité stricte. Choisissez une convention unique (strings ou enum) dans tout le projet.

Supprimer l'audit programmatiquement

Retournez AuditResult.skip() depuis la fonction décorée pour supprimer le log sur cet appel spécifique.

from audit_recorder import AuditResult

@audit(action='EXPORT', resource_type='Document')
def export_document(session, id_token):
    if already_exported:
        return AuditResult.skip(data=result, reason='Déjà audité')
    return result

---

Développeur

Linting & Formatting

ruff check . --fix   # lint + auto-fix
ruff format .        # formatage
pytest               # tests

Étapes projet

Installer en mode editable

python -m venv .venv
source .venv/bin/activate
pip install -e .[dev]

Builder

pip install --upgrade build
python -m build
# → dist/*.whl et dist/*.tar.gz

Installer le wheel dans un autre environnement

python -m venv /tmp/audit-env
source /tmp/audit-env/bin/activate
pip install dist/*.whl

---

Licence

MIT — voir LICENSE.