adstoolbox 2026.5.19

Generic functions

Alchimie Data Solutions — adsToolBox

adsToolBox est une librairie Python interne d'Alchimie Data Solutions, qui regroupe les fonctions génériques réutilisées dans les développements liés à Onyx. Elle fournit des briques homogènes pour accéder aux bases de données, industrialiser des pipelines, capturer du changement (CDC), manipuler des fichiers sur différents protocoles, et gérer les tâches transverses (logs, chrono, environnement, mails, Git, Odoo, Google Calendar).

Dépôt privé — ce repository est réservé aux employés d'Alchimie Data Solutions. Le package est toutefois publié publiquement sur PyPI sous le nom adstoolbox, et un dépôt d'exemples publics est disponible : AlchimieDataSolutions/DemoPy.

• Nom du package : adstoolbox
• Module Python : adsToolBox
• Versioning : calendaire (YYYY.MM.DD) — voir pyproject.toml
• Python : >= 3.9.2
• Licence : MIT

Sommaire

• Fonctionnalités
• Installation
• Démarrage rapide
• Structure du dépôt
• Tests
• Développement
• Dépendances principales
• Auteurs
• Licence

Fonctionnalités

Tous les symboles ci-dessous sont exposés directement depuis adsToolBox (voir adsToolBox/__init__.py).

Bases de données et pipelines

Symbole	Rôle
DataFactory	Classe abstraite commune : connect, sql_query, sql_exec, sql_scalaire, insert, insert_many, insert_bulk, upsert, upsert_many, upsert_bulk, find_text_anywhere…
DbMssql	Implémentation SQL Server (driver pymssql).
DbMysql	Implémentation MySQL (driver pymysql).
DbPgsql	Implémentation PostgreSQL (driver psycopg2).
Pipeline	Orchestration d'un transfert source → destination avec batch, déduplication par hash, inférence de schéma Polars, création automatique de la table cible.
DataComparator	Compare deux sources batch par batch et produit un rapport de différences.
ChangeDataCapture	CDC déclarative (modes append, scd1, scd2, scd4) validée par JSON Schema, avec gestion staging/persistent et synchronisation côté métier.

Fichiers, infrastructure, intégrations externes

Symbole	Rôle
FileHandler	Accès fichiers multi-backends via fsspec (local, SMB, SFTP, Azure Blob) avec transfert atomique et checksum optionnel.
GitHandler	Clonage / mise à jour de dépôts Git via token (GitPython + API GitHub).
MailReader	Lecture IMAP avec décodage robuste des en-têtes et du corps (multipart).
OdooConnector	Accès XML-RPC à Odoo (get, put, …).
GoogleCalendarConnector	Lecture/écriture d'événements Google Calendar (OAuth2).

Utilitaires transverses

Symbole	Rôle
Logger	Logger unifié console / fichier / base, avec niveaux, contexte disabled() et insertion asynchrone dans une table de détails.
timer, get_timer, set_timer, now, set_timezone	Décorateur de chronométrage et helpers de temps (timezone-aware).
retry_on_failure	Décorateur de retry avec backoff et méthode de reconnexion optionnelle.
get_public_ip	Récupération de l'IP publique (utile pour pare-feux).
Env	Chargement d'un .env trouvé automatiquement dans l'arborescence parente.

Installation

Utilisation du package (public)

Le package est publié sur PyPI et installable par n'importe qui :

pip install adstoolbox

Développement (interne ADS uniquement)

L'accès aux sources est restreint aux employés d'Alchimie Data Solutions. Une fois le dépôt cloné via les accès internes, le projet est géré avec Poetry (voir pyproject.toml et poetry.lock) :

poetry install

Un requirements.txt est également maintenu pour les environnements sans Poetry :

pip install -r requirements.txt

Démarrage rapide

Connexion à une base de données

from adsToolBox import DbPgsql, Logger, Env

logger = Logger(log_level=Logger.INFO, logger_name="adsLogger")
env = Env(logger)

db = DbPgsql({
    'database': env.PG_DWH_DB,
    'user': env.PG_DWH_USER,
    'password': env.PG_DWH_PWD,
    'port': env.PG_DWH_PORT,
    'host': env.PG_DWH_HOST
}, logger)
db.connect()

generator = db.sql_query("SELECT * FROM table_test;")

for batch in generator:
    for row in batch:
        data = row

Pipeline source → destination

from adsToolBox import Pipeline

pipeline = Pipeline(
    {
        "db_source": db_src,
        "query_source": "SELECT * FROM source_table",
        "db_destination": {
            "name": "demo",
            "db": db_dst,
            "table": "destination_table",
            "cols": ["col1", "col2"],
            "cols_def": ["INT", "VARCHAR(50)"],
        },
        "operation_type": "insert",
        "insert_method": "bulk",
        "batch_size": 10_000,
    },
    logger,
)
results = pipeline.run()

print(results)

Logger et chronomètre

from adsToolBox import Logger, set_timer, timer

set_timer(state=True)

class MyJob:
    def __init__(self) -> None:
        self.logger = Logger(log_level=Logger.DEBUG)

    @timer
    def run(self) -> None:
        self.logger.info("traitement en cours")

D'autres exemples sont disponibles dans le dépôt de démo : AlchimieDataSolutions/DemoPy.

Structure du dépôt

adsGenericFunctions/
├── adsToolBox/              # Package publié
│   ├── __init__.py          # Exports publics (symboles listés ci-dessus)
│   ├── cdc.py               # ChangeDataCapture + modes SCD
│   ├── data_comparator.py   # DataComparator
│   ├── data_factory.py      # DataFactory (classe abstraite)
│   ├── db_mssql.py          # DbMssql
│   ├── db_mysql.py          # DbMysql
│   ├── db_pgsql.py          # DbPgsql
│   ├── ddl_operations.py    # Génération DDL multi-dialecte
│   ├── dml_generator.py     # Génération DML multi-dialecte
│   ├── file_handler.py      # FileHandler (fsspec)
│   ├── git_handler.py       # GitHandler
│   ├── global_config.py     # retry_on_failure, set_timer, get_public_ip
│   ├── google_calendar.py   # GoogleCalendarConnector
│   ├── load_env.py          # Env
│   ├── logger.py            # Logger
│   ├── mail_reader.py       # MailReader
│   ├── odoo.py              # OdooConnector
│   ├── pipeline.py          # Pipeline
│   └── timer.py             # timer, now, set_timezone, get_timer
├── tests/                   # Tests unitaires (mocks)
├── integration_tests/       # Tests fonctionnels (testcontainers → Docker)
├── adsGenericFunctions.py   # Point d'entrée historique
├── pyproject.toml           # Configuration Poetry + Ruff
├── pytest.ini               # Configuration pytest
└── requirements.txt         # Dépendances figées

Tests

Les tests sont répartis en deux suites, déclarées dans pytest.ini :

• tests/ — tests unitaires avec mocks (unittest.mock), sans dépendance externe.
• integration_tests/ — tests fonctionnels avec Testcontainers (SQL Server, MySQL, PostgreSQL, Samba, Azurite) ; Docker doit être disponible.

Exécuter toutes les suites

poetry run pytest

Cibler une suite

poetry run pytest tests/                 # unitaires uniquement
poetry run pytest integration_tests/     # fonctionnels uniquement
poetry run pytest tests/test_logger.py   # un fichier précis

Tests fonctionnels — prérequis

• Docker Desktop (ou équivalent) lancé et accessible.
• Les containers sont démarrés automatiquement par les fixtures (pas de setup manuel).
• Sous Windows, la variable TESTCONTAINERS_RYUK_DISABLED=true est positionnée par les tests pour éviter les problèmes de cleanup.

Développement

Linter / formatter

Le projet utilise Ruff (configuration dans pyproject.toml, select = ["ALL"] avec quelques exceptions documentées) :

poetry run ruff check .
poetry run ruff format .

Cibles configurées : adsToolBox, tests, integration_tests. Longueur de ligne : 100.

Conventions

• Python ≥ 3.9.2, typage explicite et from __future__ import annotations quand utile.
• Docstrings en français, style concis.
• Noms en snake_case pour les méthodes publiques.
• Tests unitaires obligatoires pour toute nouvelle méthode publique.

Publication

Le package est publié sur PyPI sous le nom adstoolbox. La version suit un schéma calendaire YYYY.MM.DD défini dans pyproject.toml.

poetry build
poetry publish

Dépendances principales

Les dépendances sont déclarées dans pyproject.toml (section [tool.poetry.dependencies]).

Utilisées directement dans le code

• Bases de données : SQLAlchemy, pymssql (MSSQL), psycopg2-binary (PostgreSQL), pymysql (MySQL)
• Données : polars
• Fichiers : fsspec (accès unifié local/distant)
• Validation : jsonschema (schéma CDC)
• Intégrations : GitPython, PyGithub, google-api-python-client, google-auth-oauthlib
• Environnement / divers : python-dotenv, requests, chardet
• Tests : pytest, testcontainers

Requises à l'exécution via des backends tiers

• Backends fsspec : adlfs (Azure az://), paramiko (SFTP sftp://), smbprotocol (SMB smb://)
• Fuseaux horaires Windows : tzdata (base IANA pour zoneinfo)
• Transitives pinnées : azure-core, google-api-core, protobuf (versions figées pour éviter des conflits et warnings)

Auteurs

• Olivier Siguré — [e-mail]
• Matthieu Vannin — [e-mail]
• Antoine Ducoulombier — [e-mail]
• Pierre Baux — [e-mail]

Licence

Distribué sous licence MIT.