Metadata-Version: 2.4
Name: audit-recorder
Version: 0.4.1
Summary: Generic audit library for applications
Author: audit-recorder contributors
License: MIT License
        
        Copyright (c) 2026 audit-recorder contributors
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Keywords: audit,sqlalchemy,logging,trail
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: ruff; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-anyio; extra == "dev"
Requires-Dist: anyio[trio]; extra == "dev"
Dynamic: license-file


# 🔐 audit-recorder

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

[![Python Version](https://img.shields.io/badge/python-3.11+-blue)](https://www.python.org/downloads/)
[![SQLAlchemy](https://img.shields.io/badge/sqlalchemy-2.0+-green)](https://www.sqlalchemy.org/)
[![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE)

## ✨ 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
- 🔑 **Clés composées**: Support natif des tables à index multiples

## 🚀 Installation

```bash
pip install audit-recorder
```

---

## 📖 Quick Start

### 1️⃣ Configuration simple

```python
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

```python
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

```python
from audit_recorder import audit, transactional

@transactional  # Commit/rollback auto
@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
    # ✅ 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'`)

#### `id_param` — capture des snapshots avant/après

`id_param` indique à `@audit` comment récupérer l'ID de la ressource **avant** l'appel métier, ce qui permet de charger le snapshot `old_values`.

| Cas | `id_param` | `old_values` | `new_values` |
|-----|-----------|-------------|-------------|
| CREATE — retourne l'entité avec `id` | omis | — | ✅ résolu depuis le retour |
| CREATE — retourne un DTO différent | `'payload.id'` | — | ✅ |
| UPDATE / DELETE / action custom | `'article_id'` | ✅ | ✅ |
| Clé composite | `['user_id', 'role_id']` | ✅ | ✅ |

```python
# Chemin simple vers un argument
@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='APPROVE', resource_type='DemandeAcces', id_param='db_demande.id')
def approve(session, db_demande: DemandeAcces, id_token): ...

# Liste de chemins (clé composite)
@audit(action='UPDATE', resource_type='UserRole', id_param=['user_id', 'role_id'])
def update_user_role(session, user_id: int, role_id: int, id_token): ...
# → resource_id stocké : {"user_id": 1, "role_id": 42}

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

### 4️⃣ Consulter les logs

```python
from audit_recorder import service

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

# Clé composite
logs = service.query_logs(
    db,
    resource_type='UserRole',
    resource_id={'user_id': '1', 'role_id': '42'},
)

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:**
```python
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:**
```python
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)`

```python
@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 :

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

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

### Avec loader personnalisé

```python
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 snapshot de relations (`snapshot=`)

`snapshot=` génère automatiquement un loader qui charge l'entité avec une **session fraîche** (évite l'identity map) et applique les options SQLAlchemy fournies. `noload('*')` est ajouté automatiquement pour exclure toutes les relations non listées.

C'est l'approche recommandée quand vous souhaitez inclure des relations dans le snapshot sans écrire de loader manuellement.

```python
from sqlalchemy.orm import selectinload
from audit_recorder import audit_entity

@audit_entity(snapshot=lambda cls: [
    selectinload(cls.service).load_only(Service.nom),
    selectinload(cls.compte).load_only(Compte.prenom, Compte.nom, Compte.email),
    selectinload(cls.tags),   # sans load_only = toutes les colonnes
])
class Demande(Base):
    __tablename__ = "demandes"
    id = Column(Integer, primary_key=True)
```

Le `lambda cls: [...]` (au lieu d'une liste directe) est recommandé pour éviter les problèmes de **forward references** : les attributs de la classe ne sont résolus qu'au **premier appel** du loader, après que tous les mappers SQLAlchemy sont initialisés.

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

### Avec resource_type custom

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

> 
⚠️ Sans `resource_type` explicite, la clé enregistrée est le **nom de la classe** (`cls.__name__`), sans underscore.
`DemandeAcces` → clé `"DEMANDEACCES"`, ce qui ne correspond pas à `"DEMANDE_ACCES"`.
Si vous utilisez des enums ou des noms avec underscores, passez `resource_type` explicitement aux deux décorateurs.

### 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` :

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

### Avec clé composite (`identifier` liste)

Pour les tables avec plusieurs colonnes de clé primaire (index composite), passez une liste de noms d'attributs :

```python
@audit_entity(identifier=['user_id', 'role_id'])
class UserRole(Base):
    __tablename__ = "user_roles"
    user_id = Column(Integer, primary_key=True)
    role_id = Column(Integer, primary_key=True)
```

Le `resource_id` sera automatiquement stocké comme dict : `{"user_id": 1, "role_id": 42}`.

Compatible avec `@audit(id_param=['user_id', 'role_id'])` :

```python
@audit(action='UPDATE', resource_type='UserRole', id_param=['user_id', 'role_id'])
def update_user_role(session, user_id: int, role_id: int, id_token): ...
```

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

```python
@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".

```python
@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

```python
# 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

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

### Tout désactiver

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

### Imbriqué (fusion)

```python
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

### Filtres disponibles

| Paramètre | Type | Description |
|-----------|------|-------------|
| `resource_type` | `str \| list[str]` | Type(s) de ressource |
| `resource_id` | `dict` | Identifiant exact (clé simple : `{"id": "42"}`, clé composite : `{"user_id": "1", "role_id": "42"}`) |
| `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) |
| `populate` | `Callable` | Enrichissement des DTOs après chargement |

```python
from audit_recorder import service

# Filtre simple
logs = service.query_logs(db, resource_type='Article', action='UPDATE')

# resource_type et action acceptent aussi une liste
logs = service.query_logs(
    db,
    resource_type=['Article', 'Comment'],
    action=['CREATE', 'UPDATE', 'DELETE'],
    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} — Pages: {logs.pages}")
for log in logs.results:
    print(log)
```

### Query par ID

```python
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.

```python
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` :

```python
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.

### Filtrer sur les `extra_fields` ou `new_values` (filtres JSON)

`extra_fields_filter` et `new_values_filter` permettent de filtrer en SQL sur les colonnes JSON `extra_fields` et `new_values`. La pagination reste entièrement côté base.

Le mode de comparaison est configurable **indépendamment** pour chaque filtre via `extra_fields_filter_mode` et `new_values_filter_mode` :

| Mode | Comportement |
|---|---|
| `'ilike'` (défaut) | Recherche partielle, insensible à la casse (`ILIKE '%value%'`) |
| `'exact'` | Égalité stricte, insensible à la casse |

```python
# Recherche partielle sur extra_fields (défaut)
logs = service.query_logs(
    db,
    extra_fields_filter={'role': 'adm'},              # correspond à "admin", "superadmin"…
)

# Égalité stricte sur new_values
logs = service.query_logs(
    db,
    resource_type='DemandeAcces',
    new_values_filter={'status': 'approved'},
    new_values_filter_mode='exact',
)

# Modes différents sur chaque filtre
logs = service.query_logs(
    db,
    extra_fields_filter={'role': 'adm'},
    extra_fields_filter_mode='ilike',
    new_values_filter={'status': 'approved'},
    new_values_filter_mode='exact',
)
```

### Structure d'un log

```python
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)     # dict | None  ex: {"id": "42"} ou {"user_id": "1", "role_id": "42"}
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 :

```python
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` :

```python
# 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 :

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

### Schema

```sql
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 JSON,        -- {"id": "42"} ou {"user_id": "1", "role_id": "42"}
  
  -- SNAPSHOTS (changements)
  old_values JSON,
  new_values JSON,
  changes JSON,
  
  -- QUAND
  created_at DATETIME NOT NULL,
  
  -- EXTRA
  extra_fields JSON
);
```

---

## 🧪 Testing

```python
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

```bash
# 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`](audit_recorder/audit_decorator.py) - Décorateur `@audit`
- [`config.py`](audit_recorder/config.py) - Configuration centralisée
- [`entity.py`](audit_recorder/entity.py) - Décorateur `@audit_entity`
- [`transactional_decorator.py`](audit_recorder/transactional_decorator.py) - Décorateur `@transactional`
- [`serializers.py`](audit_recorder/serializers.py) - Registry de sérializers
- [`context.py`](audit_recorder/context.py) - Context vars et controls
- [`models.py`](audit_recorder/models.py) - Modèles SQLAlchemy et Pydantic
- [`service.py`](audit_recorder/service.py) - Requêtes et logs
- [`resolver.py`](audit_recorder/resolver.py) - Résolution d'entités

---

## 📄 Licence

MIT License - voir [LICENSE](LICENSE) pour les détails.

---
