Noria

SQL-first data pipeline framework — Jinja2 templating, multi-base de données, monitoring intégré.

"Noria" — la roue hydraulique romaine qui fait circuler l'eau avec précision et régularité.
Ici, vos données circulent avec la même fiabilité.

---

Pourquoi Noria ?

Besoin	Solution Noria
Écrire du SQL, pas du Python ORM	Templates Jinja2 directement dans vos fichiers .sql
Changer de BDD sans tout réécrire	Macros cross-DB compilées selon le dialecte cible
Orchestrer des dépendances entre requêtes	DAG déclaratif en YAML, exécution parallèle
Voir ce qui se passe	Dashboard de monitoring intégré, historique SQLite
Démarrer en 2 minutes	pip install noria && noria init mon_projet && noria run

---

Installation

pip install noria

# Avec support PostgreSQL
pip install noria[postgres]

---

Démarrage rapide

# 1. Créer un projet
noria init mon_pipeline
cd mon_pipeline

# 2. Vérifier la configuration
noria validate

# 3. Voir le SQL compilé sans exécuter
noria compile

# 4. Lancer le pipeline
noria run

# 5. Ouvrir le dashboard de monitoring
noria monitor

---

Structure d'un projet Noria

mon_pipeline/
├── noria.yml                  # Configuration du pipeline
├── sql/
│   ├── macros/                # Macros Jinja2 réutilisables
│   │   └── utils.sql.j2
│   ├── 01_clients.sql         # Requêtes SQL avec templates Jinja2
│   └── 02_rapport.sql
├── params/
│   └── prod.yml               # Paramètres par environnement
└── .noria/
    ├── data.db                # Base DuckDB locale (développement)
    └── tracking.db            # Historique des exécutions

---

Configuration — noria.yml

name: mon_pipeline
description: "Pipeline de données clients"

database:
  dialect: duckdb          # duckdb | postgresql
  database: .noria/data.db

params:
  date_min: "2024-01-01"
  schema: main

monitoring:
  enabled: true
  port: 7070

tasks:
  clients_actifs:
    sql: sql/01_clients.sql
    target: "{{ params.schema }}.clients_actifs"
    write_mode: replace        # replace | append | truncate | merge
    active: true
    description: "Clients avec au moins une commande"

  rapport:
    sql: sql/02_rapport.sql
    target: "{{ params.schema }}.rapport"
    write_mode: replace
    depends_on:
      - clients_actifs          # exécuté après clients_actifs
    pre_hook:
      file: sql/hooks/pre_rapport.sql
    post_hook:
      file: sql/hooks/post_rapport.sql

---

Écrire du SQL avec Jinja2

-- sql/01_clients.sql

SELECT
    {{ cast('id', 'VARCHAR') }}             AS id_client,  -- adapté au dialecte
    first_name || ' ' || last_name          AS nom,
    {{ date_trunc('month', 'created_at') }} AS mois_inscription,
    {{ coalesce('phone', "'N/A'") }}        AS telephone

FROM source_schema.customers
WHERE created_at >= '{{ params.date_min }}'

{% if params.actif_only %}
  AND actif = TRUE
{% endif %}

Macros cross-base de données

Macro Jinja2	PostgreSQL compilé	DuckDB compilé
{{ cast('id', 'TEXT') }}	id::TEXT	CAST(id AS TEXT)
{{ date_trunc('month', 'ts') }}	DATE_TRUNC('month', ts)	DATE_TRUNC('month', ts)
{{ current_timestamp() }}	NOW()	CURRENT_TIMESTAMP
{{ concat('a', 'b') }}	CONCAT(a, b)	CONCAT(a, b)
{{ coalesce('a', '0') }}	COALESCE(a, 0)	COALESCE(a, 0)
{{ safe_divide('num', 'den') }}	CASE WHEN den=0 THEN NULL...	CASE WHEN den=0 THEN NULL...
{{ to_char('ts', 'YYYY-MM') }}	TO_CHAR(ts, 'YYYY-MM')	STRFTIME(ts, 'YYYY-MM')
{{ limit(10) }}	LIMIT 10	LIMIT 10
{{ schema_table('s', 't') }}	s.t	s.t

Macros personnalisées

Créez vos propres macros dans sql/macros/*.sql.j2 :

{# sql/macros/utils.sql.j2 #}
{% macro panier_moyen(montant, quantite) %}
  {{ safe_divide(montant, quantite) }}
{% endmacro %}

Puis utilisez-les dans n'importe quel fichier SQL :

{% from 'utils.sql.j2' import panier_moyen %}
SELECT {{ panier_moyen('SUM(amount)', 'COUNT(id)') }} AS panier_moyen
FROM orders

---

Modes d'écriture (write_mode)

Mode	Comportement
replace	DROP TABLE + CREATE TABLE AS SELECT
append	INSERT INTO ... SELECT
truncate	TRUNCATE + INSERT
merge	INSERT ON CONFLICT (nécessite merge_key)
create_if_not_exists	Crée seulement si absente

---

CLI — Commandes

noria init <nom>          # Initialiser un projet
noria validate            # Valider config + DAG (sans connexion BDD)
noria compile             # Prévisualiser le SQL compilé
noria compile -t ma_tache # Compiler une seule tâche
noria run                 # Lancer tout le pipeline
noria run -t tache_a      # Lancer une tâche spécifique
noria run --dry-run       # Compiler sans exécuter
noria run --workers 8     # Parallélisme (défaut: 4)
noria run --param date_min=2024-06-01  # Surcharger un paramètre
noria tasks               # Lister les tâches et dépendances
noria status              # Voir l'historique des exécutions
noria monitor             # Ouvrir le dashboard web

---

Paramétrage par environnement

# Développement (défaut)
noria run

# Production via surcharge CLI
noria run --param schema=prod --param date_min=2023-01-01

# Variables d'environnement dans noria.yml
# database.password: ${DB_PASSWORD}

---

Hooks pre/post

tasks:
  ma_tache:
    sql: sql/ma_tache.sql
    pre_hook:
      file: sql/hooks/pre.sql     # Exécuté avant la tâche
      ignore_on_error: false
    post_hook:
      file: sql/hooks/post.sql    # Exécuté après
      ignore_on_error: true       # Erreur non bloquante

---

Monitoring

Le dashboard web est inclus par défaut :

noria monitor             # Ouvre http://localhost:7070

• Historique de toutes les exécutions
• Détail par tâche : statut, durée, lignes écrites, erreurs
• Rafraîchissement automatique (HTMX)

L'historique est stocké localement dans .noria/tracking.db (SQLite, zéro config).

---

API Python

from noria.core.pipeline import Pipeline

# Lancer le pipeline
pipeline = Pipeline("./mon_projet", params_override={"schema": "staging"})
results = pipeline.run()

# Compilation uniquement
compiled_sql = pipeline.compile(task_name="ma_tache")
print(compiled_sql["ma_tache"])

# Historique
history = pipeline.history(limit=5)

---

Ajouter un dialecte personnalisé

from noria.adapters import register_adapter
from noria.adapters.base import BaseAdapter
from sqlalchemy import create_engine

class MyAdapter(BaseAdapter):
    dialect_name = "mydb"
    
    def create_engine(self, db_config):
        return create_engine(f"mydb://...")
    
    # ... implémenter les méthodes abstraites

register_adapter("mydb", MyAdapter)

---

Tests

make install-dev
make test
make test-cov   # avec rapport de couverture HTML

---

Contribuer

1. Fork le dépôt
2. make install-dev
3. Créer une branche feature
4. make test + make lint avant de soumettre

---

Licence

MIT © Noria Contributors