Metadata-Version: 2.4
Name: audit-recorder
Version: 0.5.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>=1.4.42
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-1.4.42+-green)](https://www.sqlalchemy.org/)
[![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE)

---

## 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](QUICKSTART.md)**

---

## Données enregistrées par défaut

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

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

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

---

## Configuration

### 1. Créer la table

**Sans Alembic (dev / nouveau projet)**

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

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

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

### 2. Initialiser `AuditConfig`

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

```python
# 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': 'alice@example.com', '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`)

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

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

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

### Étapes projet

**Installer en mode editable**

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

**Builder**

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

**Installer le wheel dans un autre environnement**

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

---

## Licence

MIT — voir [LICENSE](LICENSE).
