API Pipelines

Composer et exécuter des DAG d'étapes de scraping, enrichissement et transformation sous /api/pipelines.

API Pipelines

Un pipeline est un graphe orienté acyclique (DAG) de nodes qui enchaîne des étapes de scraping, enrichissement et transformation. Soumettre un pipeline lance les jobs racines de façon synchrone ; les jobs en aval sont engendrés à mesure que chaque prédécesseur atteint done.

Tous les endpoints sont sous /api/pipelines et requièrent une session authentifiée. Les routes mutantes requièrent en plus un compte actif (non suspendu). Erreurs génériques : 401 pas de session, 403 non propriétaire / compte suspendu, 404 pipeline ou node inconnu ; les causes spécifiques sont listées en ligne.

Voir aussi : Orchestration de pipelines et Module filter.

Forme du graphe

Le document definition décrit le DAG. Les edges sont explicites et référencent des ids de node ; ils ne sont pas inférés d'un champ inputs par node.

{
  "nodes": [
    {"id": "n1", "type": "scrap",  "config": {"queries": ["dentist"], "zones": ["Paris"]}, "x": 100, "y": 100},
    {"id": "n2", "type": "emails", "config": {"mode": "normal"}, "x": 320, "y": 100},
    {"id": "n3", "type": "verify", "config": {}, "x": 540, "y": 100}
  ],
  "edges": [
    {"id": "e1", "from": "n1", "to": "n2"},
    {"id": "e2", "from": "n2", "to": "n3"}
  ]
}

Types de node

filter et sort préservent le type amont ; la compatibilité de types est résolue en remontant jusqu'au premier ancêtre non-passthrough.

verify lit la colonne email : il se branche aussi sur toute sortie pois dès que cette colonne existe en amont, parce qu'un bloc placé plus haut dans la chaîne la produit ou parce que la chaîne part d'un import. L'ordre des enrichissements entre emails et verify est libre, par exemple scrap → emails → legal_data → verify.

Règles de validation

Le serveur rejette une définition avec HTTP 400 si l'une des conditions suivantes est vraie :

Les racines doivent être scrap, sirene, ou import. Le fan-out (un node alimentant plusieurs successeurs) est autorisé ; le fan-in ne l'est pas.

POST /api/pipelines/validate

Dit si une définition serait acceptée par POST /api/pipelines, sans rien créer ni lancer. Fonctionne avec une session ou une clé API.

Les contrôles sont ceux de la création, dans cet ordre :

1. Structure (reason: "definition") : nodes, liaisons et config, voir Règles de validation.
2. Zones (reason: "zones") : le pays et chaque zone des nodes scrap sont résolus exactement comme au lancement. Le message nomme la zone refusée, le pays, et une suggestion quand il y en a une.
3. Coût (reason: "cost") : le pire cas de l'estimation EF, comparé au seuil du compte.
4. Place pour démarrer (reason: "busy") : les jobs racines doivent tenir dans les 5 jobs simultanés du compte. Ce refus est passager : attendre la fin d'un job, ou en annuler un.

Tester avec validate avant de créer. Une définition qui passe est acceptée par POST /api/pipelines tant que le compte garde la place pour ses jobs racines. Limité à 120 appels par 10 minutes par compte (voir Limites).

Corps de requête

{ "definition": { "nodes": [...], "edges": [...] }, "schema_version": 1 }

structure_only: true (optionnel) s'arrête à l'étape 1. L'éditeur s'en sert pour charger une définition importée, corrigée ensuite à l'écran.

Réponse 200 : acceptée

{
  "ok": true,
  "definition": { "nodes": [...], "edges": [...] },
  "summary": { "n_nodes": 3, "n_edges": 2, "types": ["scrap", "emails", "verify"] }
}

Réponse 200 : refusée (toujours HTTP 200, avec ok: false)

{ "ok": false, "reason": "zones", "error": "Zone non reconnue pour le pays France (FRA) : « Pariss » (suggestion : « Paris »). Corriger la zone, puis relancer." }

POST /api/pipelines

Crée un pipeline et lance ses jobs racines.

Corps de requête

{
  "name": "Dentistes Paris",
  "definition": { "nodes": [...], "edges": [...] },
  "low_priority": false
}

name est optionnel (≤ 120 caractères, défaut "Pipeline").

low_priority est optionnel (défaut false). Avec true, la pipeline est non prioritaire. Ses jobs passent alors après les autres dans la file d'attente. Une pipeline non prioritaire coûte moins cher.

Réponse 201

{
  "id": "f1a2…-uuid",
  "status": "running",
  "initial_jobs": ["job_abc", "job_def"]
}

Un refus ne laisse rien derrière lui. Tout ce qui peut refuser (structure, zones, coût, place dans les 5 jobs simultanés) est contrôlé avant toute écriture. Une demande refusée renvoie 400 avec {"detail": "..."} et ne crée ni pipeline ni job : aucune pipeline ne reste en failed à nettoyer. Le message donne la cause et la marche à suivre, par exemple :

• Trop de jobs en cours : 5 en attente ou en cours sur ce compte, 5 au maximum, et cette pipeline en démarre 1. Attendre la fin d'un job ou en annuler un, puis relancer.
• Zone non reconnue pour le pays France (FRA) : « Pariss » (suggestion : « Paris »). Corriger la zone, puis relancer.
• Coût estimé trop élevé : jusqu'à 7.5 EF pour cette pipeline, pour un seuil de 5 EF sur ce compte. Réduire les zones, les requêtes ou les étapes, ou relever le seuil dans Réglages.

Les appelants par clé API reçoivent ces messages en anglais.

429 : quota de création atteint (30 créations par 10 minutes et 300 par 24 heures par compte, jobs et pipelines ensemble, tentatives refusées comprises). Attendre l'en-tête Retry-After. Voir Limites.

Pour éviter les refus, contrôler d'abord la définition avec POST /api/pipelines/validate.

GET /api/pipelines

Liste les pipelines de l'appelant (plus récents d'abord, plafonné à 50).

Réponse 200

[
  {
    "id": "f1a2…",
    "name": "Dentistes Paris",
    "status": "running",
    "created_at": "2026-05-27 10:14:02",
    "completed_at": null,
    "nodes_count": 3
  }
]

status prend l'une des valeurs pending | running | done | failed | cancelled. nodes_count est dérivé de la définition stockée.

GET /api/pipelines/{id}

Retourne un pipeline avec sa définition et les jobs engendrés jusqu'ici.

Réponse 200

{
  "id": "f1a2…",
  "user_id": 42,
  "name": "Dentistes Paris",
  "definition": { "nodes": [...], "edges": [...] },
  "status": "running",
  "created_at": "2026-05-27 10:14:02",
  "completed_at": null,
  "progress_pct": 42,
  "jobs": [
    {
      "id": "job_abc",
      "job_type": "scrap",
      "status": "done",
      "pipeline_node_id": "n1",
      "results_count": 187,
      "error_message": null,
      "created_at": "2026-05-27 10:14:02",
      "completed_at": "2026-05-27 10:18:55"
    }
  ],
  "output_job": {
    "id": "job_xyz",
    "job_type": "verify_emails",
    "results_count": 142,
    "status": "done",
    "download_available": true
  }
}

output_job est le jeu de données final du pipeline : la sortie de l'étape la plus en aval ayant réellement produit des lignes (un pipeline filtre/réduit, il n'additionne pas ; output_job.results_count correspond donc au compteur affiché, pas à la somme des étapes). On le télécharge via GET /api/jobs/{id}/download avec output_job.id, en csv / xlsx / json. Il vaut null tant que le pipeline n'a rien produit de téléchargeable, et download_available indique si un CSV (final ou partiel : donc valable aussi pour un pipeline en cours/arrêté) est encore présent sur disque et non expiré.

PATCH /api/pipelines/{id}

Non implémenté. L'API actuelle n'expose pas de mutation du graphe après création ; cloner le pipeline en relançant POST /api/pipelines avec une définition mise à jour. Retourne 405 Method Not Allowed.

DELETE /api/pipelines/{id}

Non implémenté. Les pipelines sont immuables une fois créés ; la suppression sera ajoutée lorsque la politique de rétention sera définie. Retourne 405 Method Not Allowed.

POST /api/pipelines/{id}/cancel

Arrête le pipeline entier : l'étape en cours est stoppée et les étapes restantes ne sont pas lancées. Les résultats partiels déjà extraits restent téléchargeables sur chaque job concerné. À distinguer de l'arrêt d'une seule étape (POST /api/jobs/{id}/cancel), qui met la pipeline en pause sur cette étape quand le job avait démarré, et la clôt en cancelled quand il n'avait jamais démarré (voir le résumé du cycle de vie).

Le pipeline passe en statut cancelled. Renvoie 400 si le pipeline est déjà terminal (done, failed, cancelled).

Réponse : 200 OK, {"ok": true}.

POST /api/pipelines/{id}/nodes/{node_id}/continue

« Continuer avec les résultats ». Quand une étape de scraping a été arrêtée (assez de résultats) ou a crashé, la pipeline se met en pause sur cette étape (les suivantes ne démarrent pas automatiquement). Cet endpoint relance la chaîne à partir de cette étape en consommant ses résultats partiels : il crée le(s) job(s) de l'étape suivante.

Conditions : le dernier job du node est cancelled ou failed, et la pipeline n'est pas terminale. Si l'étape n'a produit aucune ligne, la branche est court-circuitée (pas de successeur créé). Renvoie 400 si l'étape n'est pas dans un état poursuivable.

Réponse : 200 OK, {"ok": true}.

POST /api/pipelines/{id}/run

Non implémenté. Les pipelines démarrent automatiquement à la création via POST /api/pipelines ; aucun endpoint de run séparé. Pour ré-exécuter un graphe, le re-poster comme nouveau pipeline.

GET /api/pipelines/{id}/nodes/{node_id}/input-columns

Inspecte le schéma du CSV qui alimentera un node donné. Utile pour construire des UIs de filtre.

Comportement. L'endpoint localise le job prédécesseur le plus récent du node. Si le prédécesseur n'est pas encore done, la réponse porte une liste columns vide et un code reason. Sinon le CSV de sortie est lu (jusqu'à 5000 lignes) et chaque colonne est profilée pour type, taux de remplissage et exemples.

Réponse 200 : prédécesseur terminé

{
  "columns": [
    {
      "name": "telephone",
      "type": "phone",
      "fill_rate": 0.92,
      "sample_values": ["+33 1 23 45 67 89", "0612345678"],
      "distinct_count": null
    },
    {
      "name": "categorie",
      "type": "category",
      "fill_rate": 1.0,
      "sample_values": ["dentiste", "orthodontiste"],
      "distinct_count": 4,
      "distinct_values": ["dentiste", "endodontiste", "orthodontiste", "stomatologue"]
    }
  ],
  "row_count": 187,
  "predecessor_job_id": "job_abc"
}

type prend l'une des valeurs phone | email | url | number | category | text. Une colonne est étiquetée category uniquement si elle a entre 1 et 200 valeurs non vides distinctes ; sinon elle retombe à text. Un verdict typé requiert ≥ 80% des valeurs non vides correspondant au pattern.

Réponse 200 : pas d'entrée utilisable

{ "columns": [], "reason": "no_predecessor" }

POST /api/pipelines/{id}/nodes/{node_id}/filter-preview

Applique un jeu de règles de filtre en mémoire contre le CSV amont et renvoie le nombre de correspondances plus un petit échantillon. Aucun job n'est créé ; aucun état n'est muté.

Le node cible doit être de type filter. Le corps utilise la même forme rules que les nodes filter persistent dans leur config.rules ; les previews sont calculés par la même fonction que le worker utilise à l'exécution, donc le compte fait autorité pour les données inspectées.

Corps de requête

{
  "rules": {
    "logic": "AND",
    "conditions": [
      {"column": "fill_rate", "op": ">=", "value": 0.5},
      {"column": "categorie", "op": "in", "value": ["dentiste", "orthodontiste"]}
    ]
  }
}

La grammaire exacte des règles est définie par le module filter (voir Module filter).

Réponse 200

{
  "total": 187,
  "matched": 73,
  "samples": [
    {"nom": "Cabinet Dupont", "telephone": "0123456789", "categorie": "dentiste"}
  ],
  "predecessor_job_id": "job_abc",
  "fieldnames": ["nom", "telephone", "categorie", "site_web"],
  "capped": false
}

samples contient jusqu'à 5 lignes correspondantes avec les champs vides retirés. capped vaut true quand le CSV amont a dépassé la limite preview de 5000 lignes : dans ce cas total ne reflète que la fenêtre inspectée, mais le ratio matched/total reste représentatif.

Quand le prédécesseur n'est pas prêt, la réponse est la même squelette {total, matched, samples, reason} avec tous les compteurs à 0. Les codes reason possibles miroitent l'endpoint input-columns : no_predecessor, no_data_yet, no_csv_found.

Causes spécifiques : 400 node cible n'est pas de type filter, ou application de règle a levé une erreur ; 500 CSV non lisible.

Résumé du cycle de vie

1. POST /api/pipelines valide le graphe, les zones, le coût et la place dans le quota de jobs simultanés, puis persiste le pipeline en running et engendre un job par node racine. Une demande refusée ne persiste rien.
2. Quand un job se termine normalement (done), le worker lit son CSV, transforme les lignes pour le type d'entrée du successeur et crée le job suivant. Une sortie vide court-circuite la branche.
3. Quand un job est arrêté (cancelled) ou crashe (failed), la pipeline se met en pause sur cette étape : aucune suite automatique. L'utilisateur reprend explicitement via POST /api/pipelines/{id}/nodes/{node_id}/continue (« Continuer avec les résultats »), qui crée le job suivant à partir du CSV partiel. Exception : un job annulé alors qu'il n'avait jamais démarré (pending, aucun résultat) clôt la pipeline en cancelled, sauf si une autre de ses étapes est encore en attente ou en cours.
4. La pipeline est finalisée en done quand toutes les étapes atteignables sont terminales et résolues. Un arrêt global (POST /api/pipelines/{id}/cancel) la finalise en cancelled. Un job expired la finalise en failed. Une étape cancelled/failed non poursuivie ne finalise rien : la pipeline reste running (en pause) jusqu'à un « Continuer » ou un arrêt global.