Créer un adaptateur personnalisé

Noria expose une API de plugin qui permet d'ajouter un nouvel adaptateur sans modifier le framework.

1. Créer la classe adaptateur

Héritez de noria.adapters.base.BaseAdapter et implémentez toutes les méthodes abstraites :

# my_pkg/my_adapter.py
from noria.adapters.base import BaseAdapter
from sqlalchemy import Engine, create_engine
from typing import Any

class ClickHouseAdapter(BaseAdapter):
    dialect_name = "clickhouse"

    def create_engine(self, db_config: Any) -> Engine:
        url = f"clickhouse+native://{db_config.host}:{db_config.port}/{db_config.database}"
        return create_engine(url)

    def drop_table_sql(self, schema, table):
        return f"DROP TABLE IF EXISTS {self._qualified(schema, table)}"

    def create_table_as_sql(self, schema, table, select_sql):
        qt = self._qualified(schema, table)
        return f"CREATE TABLE {qt} ENGINE=MergeTree() ORDER BY tuple() AS\n{select_sql}"

    # ... implémenter les autres méthodes abstraites

    def _macro_cast(self, col, type_):
        return f"CAST({col}, '{type_}')"

2. Enregistrer l'adaptateur

from noria.adapters import register_adapter
from my_pkg.my_adapter import ClickHouseAdapter

register_adapter("clickhouse", ClickHouseAdapter)

Appelez register_adapter avant d'instancier un Pipeline, par exemple dans un hook d'entrée ou un script d'initialisation.

3. Utiliser dans noria.yml

database:
  dialect: clickhouse
  host: localhost
  port: 9000
  database: default

Méthodes abstraites requises

Méthode	Description
`create_engine(db_config)`	Retourne un `sqlalchemy.Engine`
`drop_table_sql(schema, table)`	DROP TABLE IF EXISTS
`create_table_as_sql(schema, table, select)`	CREATE TABLE AS SELECT
`truncate_sql(schema, table)`	TRUNCATE
`insert_into_sql(schema, table, select)`	INSERT INTO … SELECT
`merge_sql(schema, table, select, keys)`	Upsert par clé
`incremental_sql(schema, table, select, key_col)`	Refresh partition
`scd2_sql(schema, table, select, keys)`	SCD Type 2
`update_sql(schema, table, select, keys, cols)`	UPDATE ciblé
`table_exists_sql(schema, table)`	Vérifier existence table
`row_count_sql(schema, table)`	Compter les lignes