audit-recorder 0.1.0

Generic audit library for applications

Cette librairie permet d'auditer les actions sur un projet et mets également à disposition une UI intégrée.

Structure complète du projet

project-audit/
├── audit_recorder/
│   ├── __init__.py
│   ├── context.py
│   ├── audit_decorator.py
│   ├── transactional_decorator.py
│   ├── service.py
│   ├── models.py
│   ├── resolver.py
│   └── result.py
├── tests/
├── LICENSE
├── README.md
└── pyproject.toml

Formatting with ruff

ruff check . --fix ruff format .

Tests

pytest

Builder la lib

python -m pip install --upgrade build
python -m build

Le répertoire dist contiendra un fichier .whl et un fichier .tar.gz.

Installer la distribution sur un projet (en local)

pip install /path/to/project-audit/dist/audit_recorder-0.1.0-py3-none-any.whl
pip show audit-recorder

Installer la distribution sur un projet (en mode dev)

pip install -e /home/you/dev/project-audit

Executer les tests unitaires

pytest -v

Intégration de la base de données (sqlalchemy et alembic)

Importer et ajouter la metadata dans le fichier de configuration :

from audit_recorder import models as audit_recorder_models

target_metadata = [app.models.model.Base.metadata, audit_recorder_models.Base.metadata]

Générer et executer le script de mise à jour, sous Alembic :

alembic revision --autogenerate -m "create_audit_logs_from_audit_recorder"
alembic upgrade head

Intégration du service

Configuration

Créer un fichier audit_setup.py. Celui-ci condiendra les requêtes qui seront executées avant et après chaque action, afin d'historiser les données et les changements. Les actions et les types de resources sont à la charge de chaque projet. TODO : héritage

Créer autant de query et de register que vous avez de type de resource. Exemple :

from audit_recorder.resolver import resolver
from sqlalchemy.orm import Session
from app.models.model import Compte

class AuditResourceEnum(enum.Enum):
    COMPTE = "COMPTE"

class AuditActionEnum(enum.Enum):
    APPROVE = "APPROVE"
    REFUSE = "REFUSE"

def get_compte(session: Session, id: int) -> Compte | None:
    return session.query(Compte).get(id)

def audit_setup():
    resolver.register(AuditResourceEnum.COMPTE.name, get_compte, Compte.id)

La méthode register accepte trois paramètres :

• resource_type : le nom de la ressource
• loader : la fonction de chargement de la ressource, signature (session: Session, resource_id: Any) -> Any
• identifier : le champ SQLAlchemy utilisé comme identifiant de la ressource (ex. Compte.id, Compte.uuid)

L'identifier sert à extraire automatiquement l'id depuis l'objet retourné par la fonction décorée lorsque id_param n'est pas renseigné, notamment dans le cas d'une création.

Appeler la méthode audit_setup() dans le context manager :

from app.audit_setup import audit_setup

@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    ...
    audit_setup()
    yield

Décorateur @audit

La lib mets à disposition un décorateur afin d'auditer une action. Exemple d'utilisation :

from audit_recorder.decorators.audit_decorator import audit

@audit(action="UPDATE", resource_type="COMPTE", id_param="compteDTO.id")
async def update_compte(compteDTO: CompteDTO, id_token: KeycloakIDToken, session: Session) -> CompteDTO:

Si id_param n'est pas renseigné, l'identifiant sera extrait automatiquement depuis l'objet retourné par la fonction, en utilisant l'identifier enregistré dans le resolver (via resolver.register). Exemple :

@audit(action="CREATE", resource_type="COMPTE")  # pas d'id_param
async def create_compte(compteDTO: CompteDTO, id_token: KeycloakIDToken, session: Session) -> Compte:
    ...
    return db_compte  # db_compte.id sera extrait via Compte.id déclaré dans le resolver

Important: - Tous les paramètres du décorateur @audit doivent être passés en nommés (action=, resource_type=, id_param=) - Tous les arguments de la fonction décorée doivent également être passés en nommés lors de l'appel - id_token: KeycloakIDToken est obligatoire afin de pouvoir récupérer les informations sur l'utilisateur - session: Session est obligatoire afin que les logs d'audit soient ajoutées dans la même session

Gestion de la transaction - décorateur @transactional

Le décorateur @audit ne réalise aucun commit, c'est action à la charge de la méthode appelante. Un décorateur @transactional est mis à disposition afin d'automatiser les commits, exemple :

from audit_recorder.decorators.audit_decorator import audit
from audit_recorder.decorators.transactional_decorator import transactional

@transactional
@audit(action=AuditActionEnum.CREATE.name, resource_type=AuditResourceEnum.COMPTE.name, id_param="db_compte.id")
def approve_demande_creation(...)

A noter: l'ordre est important, le l'annotation @transactional doit précéder l'annotation @audit pour la persitance des données.

Contexte

Dans certain cas, on souhaite auditer une action en fonction de l'appelant. Par exemple, auditer une approbation manuelle mais ne pas auditer une autoapprobation.

Pour cela il est possible de définir un contexte afin de désactiver un ou plusieurs audits, exemple :

from audit_recorder.context import skip_audit

with skip_audit(AuditActionEnum.APPROVE, AuditActionEnum.REJECT)
with skip_audit():
    service.approve(...)

Ignorer l'audit

L'audit est automatiquement réalisée après l'execution de la méthode annotée, sauf si celle-ci lance une exception. Il est possible d'ignorer explicitement l'audit. Exemple :

from audit_recorder.result import AuditResult

def approve(...) {
    if is_already_approve:
        return AuditResult.skip(db_compte)
}

Services

La lib mets à disposition des services afin de récupérer des données filtrées et paginées :

from audit_recorder import service

service.query_logs(...)
service.query_log_by_id(...)