sql-lineage-extractor 0.1.1

Static column-to-column SQL lineage extractor producing a dbt-manifest-like JSON graph.

sql-lineage-extractor

Extraction statique de lineage SQL colonne-à-colonne, inspirée du manifest.json de dbt — mais sans moteur d'exécution ni orchestration. L'outil n'exécute aucune requête de transformation, ne charge aucune donnée : il analyse du texte SQL et produit un graphe structuré (JSON versionné et diffable).

Le SQL analysé peut provenir indifféremment de :

• fichiers .sql ;
• notebooks Fabric / Jupyter (.ipynb) ;
• une connexion live à un endpoint SQL (Fabric SQL endpoint, SQL Server, Synapse) via une connexion DB-API générique injectée de l'extérieur.

Aucune dépendance à un SDK Fabric propriétaire. Le parsing repose sur sqlglot.

Installation

pip install -e .
# mode live (connexion DB-API, ex. pyodbc) :
pip install -e ".[live]"
# outillage de dev (tests) :
pip install -e ".[dev]"

Python ≥ 3.10.

Architecture — pipeline en 4 couches

Sources SQL (fichiers / notebooks / live)
        │  -> SQLUnit (id, dialecte, sql_text, origine)
        ▼
Parsing + lineage colonne-à-colonne (sqlglot)   -> ObjectNode | ParseError
        ▼
Construction du graphe multi-objets (depends_on_nodes)
        ▼
Sérialisation en manifest JSON (schématisé, déterministe)

Chaque couche est testable indépendamment (tests/).

Utilisation (CLI)

# Fichiers .sql
sql-lineage extract --source files     --path ./vues       --dialect tsql  --out manifest.json

# Notebooks .ipynb
sql-lineage extract --source notebooks --path ./notebooks  --dialect spark --out manifest.json

# Endpoint SQL live (nécessite l'extra [live])
sql-lineage extract --source live      --conn-string "<DSN ou chaîne ODBC>" --dialect tsql --out manifest.json

# Vérifier qu'un manifest est conforme au schéma publié
sql-lineage validate manifest.json

Options utiles : --glob pour restreindre le motif de fichiers (--glob "silver/**/*.sql").

Utilisation programmatique

from sql_lineage.sources import FileSQLSource
from sql_lineage.pipeline import run_extraction
from sql_lineage.manifest import write_manifest

nodes, errors = run_extraction(FileSQLSource("./vues", dialect="tsql"))
write_manifest("manifest.json", nodes, errors)

Pour le mode live, la connexion est fournie par l'appelant (jamais construite dans la librairie) :

import pyodbc
from sql_lineage.sources import LiveSQLEndpointSource

conn = pyodbc.connect("Driver=...;Server=...;Authentication=ActiveDirectoryInteractive;...")
source = LiveSQLEndpointSource(conn, dialect="tsql")

Schéma du manifest

Documenté et versionné dans src/sql_lineage/manifest/schema.json (schema_version en tête de fichier, actuellement 1.0). La politique d'évolution du contrat (MAJOR.MINOR, dépréciation, golden tests) est décrite dans docs/MANIFEST_SCHEMA.md. Extrait :

{
  "schema_version": "1.0",
  "generated_at": "2026-07-20T10:00:00Z",
  "nodes": {
    "silver_shipping_escale": {
      "origine": "vues-shipping/silver/escale.sql",
      "dialecte": "tsql",
      "columns": {
        "date_traitement": {
          "expression_sql": "e.date_arrivee AS date_traitement",
          "depends_on": [{"table": "bronze_navis.escale", "column": "date_arrivee"}]
        }
      },
      "depends_on_nodes": ["bronze_navis_escale"],
      "parametres_neutralises": [],
      "avertissements": []
    }
  },
  "errors": [
    {"id": "silver_finance_xyz", "origine": "...", "message": "..."}
  ]
}

Déterminisme : même entrée → même sortie. Clés triées, listes ordonnées, formatage stable — pour un diff propre en CI. Le seul champ non reproductible, generated_at, est injectable (build_manifest(..., generated_at=...)).

Export vers la BI (Power BI, etc.)

Le manifest JSON reste canonique (contrat versionné, diffable en CI). Pour la BI on en dérive une couche de service tabulaire — schéma en étoile — que Power BI consomme directement, sans avoir à « expand » du JSON imbriqué :

manifest.json  (canonique)
      │  sql-lineage export  (dérivation pure, rejouable, append par run)
      ▼
dim_run · dim_object · dim_column · fact_column_edge · object_edge · closure · errors

Table	Grain	Rôle
dim_run	1 run	dimension temps (historisation)
dim_object	1 objet	dimension (slice par layer bronze/silver/gold)
dim_column	1 colonne	dimension colonne
fact_column_edge	1 arête colonne→colonne	fait central
object_edge	1 dépendance objet directe	graphe 1-saut
closure	1 couple ancêtre→descendant	multi-saut (impact / provenance)
errors	1 erreur	suivi qualité

Toutes les tables portent un run_id : chaque run est ajouté (jamais écrasé), ce qui ouvre la dérive dans le temps, l'« as-of date » et l'audit. La table closure (fermeture transitive précalculée en Python) débloque les rapports impact (« si cette source change, quoi en aval ? ») et provenance (« d'où vient cette colonne à la source ? ») sur plusieurs niveaux — impossible en DAX récursif sur un DAG.

Backends

# CSV (un fichier par table)
sql-lineage export manifest.json --format csv       --out ./exports

# SQLite (fichier unique, lisible par Power BI via ODBC)
sql-lineage export manifest.json --format sqlite    --out lineage.db

# SQL Server / Fabric Warehouse (connexion injectée)
sql-lineage export manifest.json --format sqlserver --conn-string "<chaîne ODBC>"

Le backend sqlserver utilise l'extra [live] (pyodbc). Les backends base de données sont idempotents par run (les lignes d'un run_id existant sont remplacées) ; le CSV est append-only.

Fabric : guide d'usage complet (exécution hors Spark, connexion Warehouse via Service Principal, script d'orchestration prêt à coller) dans docs/FABRIC_USAGE.md et examples/fabric_orchestration.py.

Modélisation Power BI (indicatif)

• Relations : dim_object[node_id] 1—* fact_column_edge[target_node] ; dim_run[run_id] 1—* sur chaque table (filtre de run).
• Impact/provenance : filtrer closure[ancestor_node] (aval) ou closure[descendant_node] (amont), depth donne la distance.
• Volume faible → mode Import suffit (pas besoin de DirectQuery/DirectLake).

Conventions

Identifiant d'un objet dans un notebook

Priorité de résolution de l'id d'une cellule / d'un appel spark.sql :

1. Commentaire de convention en tête du SQL : -- lineage:id=silver_shipping_escale (valeur par défaut, ajustable si l'équipe data préfère une autre convention).
2. Nom de la variable cible en Python (df_silver_finance = spark.sql(...) → silver_finance ; le préfixe df_ est retiré).
3. Repli nom_notebook#numero_cellule — ce cas émet un avertissement (champ avertissements), il n'est jamais silencieux.

Neutralisation des f-strings

Une cellule Python spark.sql(f"... WHERE d >= '{date_execution}'") est parsée avec le module ast (jamais par regex). Chaque interpolation {...} en position valeur (>= {x}, '{x}', IN ({x})) est remplacée par le jeton neutre __PARAM__ avant transmission à sqlglot, pour ne pas casser le parsing.

Une interpolation qui occupe une ligne à elle seule est traitée comme un fragment de clause SQL (motif courant en Fabric : {'' if terminal_names is None else 'AND t.col IN ' + terminal_names}) et supprimée plutôt que tokenisée — un __PARAM__ orphelin casserait le parsing, et un prédicat WHERE/HAVING ne contribue jamais au lineage colonne.

Dans les deux cas, les fragments remplacés/supprimés sont conservés dans le champ parametres_neutralises du nœud, pour transparence.

Personnaliser les conventions (LineageConfig)

Aucune convention n'est codée en dur : le commentaire d'id, le préfixe de variable, le jeton de neutralisation et le dialecte par défaut proviennent tous d'un objet LineageConfig injectable. Les défauts reproduisent exactement le comportement décrit ci-dessus (rétro-compatible).

from sql_lineage import LineageConfig
from sql_lineage.sources import NotebookSQLSource

config = LineageConfig(
    id_comment_pattern=r"--\s*lin:id\s*=\s*(\w+)",  # 1 groupe de capture = l'id
    var_prefixes=("df_", "result_"),                 # préfixes retirés de l'id
    param_token="@@PARAM@@",
    default_dialect="spark",
)
source = NotebookSQLSource("./notebooks", config=config)

La même chose en CLI (les flags surchargent un éventuel --config) :

sql-lineage extract --source notebooks --path ./notebooks \
    --config conventions.toml \
    --var-prefix df_ --var-prefix result_ \
    --id-comment-pattern '--\s*lin:id\s*=\s*(\w+)' \
    --param-token '@@PARAM@@' \
    --out manifest.json

Fichier de conventions (.toml ou .json) — les clés peuvent aussi vivre sous [tool.sql_lineage] d'un pyproject.toml partagé :

[sql_lineage]
var_prefixes = ["df_", "result_"]
param_token = "@@PARAM@@"
default_dialect = "spark"

La convention -- lineage:id= s'applique aussi aux fichiers .sql : un commentaire présent dans le fichier l'emporte sur l'id dérivé du nom de fichier.

Cas d'erreur (attendus)

Une erreur de parsing sur un objet n'interrompt jamais le traitement des autres : elle est capturée et reportée dans errors. Cas rejetés :

• SELECT * (et t.*) sur une table physique : colonnes inconnues en analyse statique. En revanche, un SELECT * au-dessus d'un CTE ou d'une sous-requête énumérable est expansé automatiquement (y compris les cas imbriqués T.*), donc supporté.
• Colonne calculée sans alias (e.a + e.b) : pas de nom de sortie stable.
• SQL non analysable par sqlglot pour le dialecte donné.

Limites connues

• SQL procédural complexe (batchs T-SQL multi-instructions, variables, MERGE, procédures stockées) : seul le premier SELECT pertinent est analysé.
• Dialectes : testés sur tsql et spark. Les autres dialectes sqlglot fonctionnent probablement mais ne sont pas couverts par les tests — à étendre selon les besoins.
• Résolution du graphe : le rapprochement table source ↔ objet normalise les points en underscores (bronze_navis.escale ↔ bronze_navis_escale) et autorise un match sur le dernier segment. Des collisions de noms courts sont théoriquement possibles.
• Non-objectifs : pas d'exécution de requêtes, pas de génération de site de documentation (le manifest est destiné à être consommé par un outil de rendu séparé), pas de tests de qualité de données.

Tests

pytest

Aucun test ne nécessite de connexion réseau ni d'accès à un environnement Fabric réel — le mode live est couvert par une connexion mockée.

Structure du dépôt

src/sql_lineage/
  models.py              # SQLUnit, ColumnLineage, ObjectNode, ParseError, ColumnSource
  sources/               # file / notebook / live  (interface commune SQLSource)
  parsing/               # wrapper sqlglot + extraction de lineage
  graph/                 # construction des arêtes depends_on_nodes
  manifest/              # schema.json + writer (sérialisation déterministe)
  pipeline.py            # orchestration source -> parse -> graph
  cli.py                 # commandes `extract` / `validate` (click)
tests/                   # fixtures .sql / .ipynb + tests par couche