Prise en charge de SQLAlchemy

Prise en charge de SQLAlchemy et Alembic pour ClickHouse

ClickHouse Connect inclut le dialecte SQLAlchemy clickhousedb, basé sur le pilote principal. Il prend en charge SQLAlchemy 1.4.40 et les versions ultérieures, y compris SQLAlchemy 2.x, avec un accent particulier sur les requêtes Core, le DDL ClickHouse, la réflexion et les insertions ORM simples.

Installez les dépendances SQLAlchemy avec l’extra du paquet :

pip install "clickhouse-connect[sqlalchemy]"

​

Se connecter avec SQLAlchemy

Créez un moteur avec l’une ou l’autre des URL suivantes :

from sqlalchemy import create_engine, text

engine = create_engine(
    "clickhousedb://user:password@host:8123/mydb?compression=zstd"
)

with engine.connect() as conn:
    version = conn.execute(text("SELECT version()")).scalar_one()
    print(version)

Les paramètres de requête d’URL peuvent contenir des paramètres ClickHouse, des options du client ClickHouse Connect telles que compression, query_limit et des dépassements de délai, ou des options HTTP/TLS telles que ca_cert. Préfixez un paramètre ClickHouse par ch_ pour qu’il soit traité comme un paramètre serveur si nécessaire, par exemple ch_http_max_field_name_size=99999.

Consultez Arguments et paramètres de connexion pour connaître les options client disponibles.

​

Paramètres par requête

Transmettez les paramètres ClickHouse via les options d’exécution de SQLAlchemy. Les paramètres peuvent être définis au niveau du moteur, de la connexion ou de l’instruction. Une valeur définie sur l’instruction prévaut sur une valeur de connexion ou de moteur avec la même clé.

from sqlalchemy import text

stmt = text("SELECT getSetting('max_threads')").execution_options(
    settings={"max_threads": 2}
)

with engine.connect() as conn:
    value = conn.execute(stmt).scalar_one()

​

Formats de lecture par requête

Définissez les formats de lecture ClickHouse pour un moteur, une connexion ou une instruction à l’aide des options d’exécution SQLAlchemy et de query_formats. Les formats définis au niveau de l’instruction sont appliqués en premier et remplacent les keys et wildcards correspondants définis au niveau de la connexion ou du moteur.

from sqlalchemy import text

stmt = text("SELECT user_uuid FROM users").execution_options(
    query_formats={"UUID": "string"}
)

with engine.connect() as conn:
    rows = conn.execute(stmt).all()

​

Paramètres côté serveur

SQLAlchemy génère normalement des paramètres côté client. Pour utiliser les paramètres côté serveur de ClickHouse, activez-les lors de la création du moteur :

engine = create_engine(
    "clickhousedb://user:password@host:8123/mydb",
    server_side_params=True,
)

Dans ce mode, chaque valeur liée doit avoir un type SQLAlchemy compatible avec ClickHouse. Les listes IN prises en charge deviennent des paramètres ClickHouse Array typés. Le compilateur lève une CompileError lorsqu’il ne peut pas déduire un type compatible ni traiter un bind en toute sécurité.

​

Requêtes Core

Le dialecte prend en charge les requêtes SELECT de SQLAlchemy Core avec des jointures, des filtres, le tri, des clauses LIMIT et OFFSET, et DISTINCT.

from sqlalchemy import MetaData, Table, select

metadata = MetaData(schema="mydb")
users = Table("users", metadata, autoload_with=engine)
orders = Table("orders", metadata, autoload_with=engine)
events = Table("events", metadata, autoload_with=engine)

stmt = (
    select(users.c.name, orders.c.product)
    .select_from(users.join(orders, users.c.id == orders.c.user_id))
    .order_by(users.c.name)
    .limit(10)
)

with engine.connect() as conn:
    rows = conn.execute(stmt).all()

Le DELETE léger est pris en charge et nécessite une clause WHERE explicite :

from sqlalchemy import delete

stmt = delete(users).where(users.c.name.like("%temporary%"))
with engine.connect() as conn:
    conn.execute(stmt)

​

Sous-colonnes JSON

Pour une colonne déclarée ou représentée en JSON ClickHouse, utilisez des crochets pour sélectionner un segment à la fois dans le chemin d’une sous-colonne stockée :

from sqlalchemy import Column, MetaData, Table, select

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import JSON, UInt32

events = Table(
    "events",
    MetaData(),
    Column("payload", JSON),
)

request_id = events.c.payload["context"]["request"].subcolumn(
    "id",
    type_=UInt32,
)

stmt = select(
    events.c.payload["severity"].label("severity"),
    request_id.label("request_id"),
)

payload["severity"] est compilé selon la syntaxe d’identifiant pointé de ClickHouse. Chaque partie est entourée de guillemets séparément, par exemple `events`.`payload`.`severity`. Cette syntaxe lit la sous-colonne JSON stockée dans ClickHouse et n’appelle pas getSubcolumn. Chaînez [] ou .subcolumn() une fois pour chaque segment du chemin. Chaque segment doit être une chaîne non vide.

Passer type_ à .subcolumn() encapsule le chemin pointé dans un CAST SQL et affecte ce type à l’expression SQLAlchemy. Sans type_, .subcolumn("segment") se comporte comme ["segment"].

Un chemin non typé a le type Dynamic de ClickHouse. ClickHouse n’autorise pas les valeurs Dynamic directement dans ORDER BY ou GROUP BY. Passez type_ lorsqu’une sous-colonne y est utilisée.

Pour du code à typage statique, importez json_subcolumn depuis clickhouse_connect.cc_sqlalchemy. Cet assistant accepte également un segment à la fois et préserve le type de résultat Python de type_ :

from clickhouse_connect.cc_sqlalchemy import json_subcolumn

context = json_subcolumn(events.c.payload, "context")
request = json_subcolumn(context, "request")
request_id = json_subcolumn(request, "id", type_=UInt32)

Dans cet exemple, les vérificateurs de types considèrent request_id comme un ColumnElement[int].

Chaque segment est entouré de guillemets individuellement, y compris les noms contenant des espaces ou des accents graves. Les accents graves ne font pas d’un point un caractère littéral pour le traitement des chemins JSON par ClickHouse. Lorsque json_type_escape_dots_in_keys est activé, utilisez l’encodage %2E de ClickHouse pour les points littéraux dans les clés. Accédez à une clé nommée a.b avec payload["a%2Eb"], et non payload["a.b"].

​

Extensions des requêtes ClickHouse

Importez select depuis clickhouse_connect.cc_sqlalchemy pour exposer des méthodes ClickHouse typées aux outils de vérification statique des types. Le sqlalchemy.select standard propose également ces méthodes à l’exécution.

from clickhouse_connect.cc_sqlalchemy import select

stmt = (
    select(events.c.user_id, events.c.event_type)
    .final()
    .prewhere(events.c.event_date >= "2026-01-01")
    .sample(0.1)
    .limit_by([events.c.user_id], 3)
)

Les méthodes Select de ClickHouse sont :

Méthode	Fonctionnalité SQL
.final()	FINAL pour une table
.sample(value)	SAMPLE, à l’aide d’une fraction, d’un nombre de lignes ou d’une expression
.prewhere(expression)	PREWHERE ; les appels répétés sont combinés avec AND
.limit_by(columns, limit, offset=None)	LIMIT ... BY
.array_join(...)	ARRAY JOIN
.left_array_join(...)	LEFT ARRAY JOIN
.ch_join(...)	jointures ClickHouse avec les options strictness, distribution, using et cross
.cte(name, materialized=True)	WITH name AS MATERIALIZED (...)

Par exemple, un GLOBAL ANY LEFT JOIN ClickHouse peut être chaîné sans avoir à imbriquer une FromClause personnalisée :

stmt = (
    select(events.c.id, users.c.name)
    .select_from(events)
    .ch_join(
        users,
        events.c.user_id == users.c.id,
        isouter=True,
        strictness="ANY",
        distribution="GLOBAL",
    )
)

Utilisez la syntaxe explicite Lambda pour les fonctions d’ordre supérieur de ClickHouse :

from sqlalchemy import column, func

from clickhouse_connect.cc_sqlalchemy import Lambda, select

stmt = select(
    func.arrayMap(
        Lambda("x", column("x") * 2),
        events.c.metrics,
    ).label("doubled")
)

La construction standard SQLAlchemy values() est compilée vers la syntaxe de fonction de table VALUES de ClickHouse, y compris lorsqu’elle est utilisée dans une expression de table commune. La forme CTE nécessite SQLAlchemy 2.0.42 ou version ultérieure, où Values.cte() a été ajoutée.

​

CTE matérialisées

Par défaut, ClickHouse intègre une expression de table commune ; ainsi, le corps d’une CTE référencée plusieurs fois est exécuté une fois par référence. Passez materialized=True à .cte() pour générer WITH <name> AS MATERIALIZED (...), ce qui calcule le corps une seule fois :

from sqlalchemy import func

from clickhouse_connect.cc_sqlalchemy import select

ranked = (
    select(book.c.book_id, func.row_number().over(order_by=book.c.score.desc()).label("result_rank"))
    .where(book.c.genre == "sci-fi")
    .order_by(book.c.score.desc())
    .limit(100)
    .cte("ranked", materialized=True)
)

stmt = (
    select(book.c.book_id, ranked.c.result_rank)
    .select_from(book)
    .ch_join(ranked, book.c.book_id == ranked.c.book_id, strictness="ANY")
    .where(book.c.book_id.in_(select(ranked.c.book_id)))
    .execution_options(settings={"enable_materialized_cte": 1, "enable_analyzer": 1})
)

Le serveur ne matérialise la CTE que si le mot-clé est présent, si enable_materialized_cte=1 et si l’analyseur est activé. Définissez enable_materialized_cte au niveau de l’instruction, de la connexion ou du moteur, comme indiqué dans Paramètres par requête. L’analyseur est activé par défaut sur tous les serveurs prenant en charge cette fonctionnalité. Définir explicitement enable_analyzer=1 constitue donc une mesure de précaution. enable_materialized_cte est un paramètre ClickHouse expérimental. Avec enable_materialized_cte=0 ou enable_analyzer=0, la requête aboutit et renvoie les mêmes lignes. ClickHouse ignore silencieusement MATERIALIZED et intègre de nouveau la CTE, de sorte qu’un paramètre oublié dégrade les performances sans générer d’erreur. Les CTE matérialisées nécessitent ClickHouse 26.3 ou une version ultérieure. Les serveurs plus anciens rejettent le mot-clé avec une erreur de syntaxe.

Pour une instruction construite avec le sqlalchemy.select standard, utilisez plutôt cte() au niveau du module. Cette fonction prend l’instruction comme premier argument et correspond par ailleurs à Select.cte() :

from sqlalchemy import select as sa_select

from clickhouse_connect.cc_sqlalchemy import cte

ranked = cte(sa_select(book.c.book_id), "ranked", materialized=True)

Le mot-clé est généré uniquement avec le dialecte ClickHouse. Une instruction partagée avec un autre backend y est donc compilée sans modification.

ClickHouse ne prend pas en charge les CTE matérialisées récursives. Les helpers SQLAlchemy lèvent une ValueError lorsque recursive=True et materialized=True sont tous deux définis.

​

DDL et introspection

ClickHouse Connect fournit les types de données ClickHouse, les moteurs de table, les structures de dictionnaire, le DDL des bases de données et l’introspection des tables.

import sqlalchemy as db
from sqlalchemy import MetaData

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import DateTime64, String, UInt32
from clickhouse_connect.cc_sqlalchemy.ddl.custom import CreateDatabase
from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree

with engine.connect() as conn:
    conn.execute(CreateDatabase("example_db", exists_ok=True))

    metadata = MetaData(schema="example_db")
    events = db.Table(
        "events",
        metadata,
        db.Column("id", UInt32, primary_key=True),
        db.Column("user", String),
        db.Column("created_at", DateTime64(3)),
        MergeTree(order_by="id"),
    )
    events.create(conn)

    reflected = db.Table("events", MetaData(schema="example_db"), autoload_with=conn)
    assert reflected.engine is not None

Les colonnes introspectées utilisent server_default pour les expressions DEFAULT, ainsi que des attributs propres au dialecte tels que clickhouse_codec, clickhouse_ttl, clickhouse_materialized et clickhouse_alias, lorsqu’ils sont présents.

Les arguments de clé MergeTree tels que order_by, partition_by, primary_key, sample_by et ttl acceptent des colonnes SQLAlchemy, des expressions SQL ainsi que de simples chaînes de caractères.

​

Insertions et utilisation de l’ORM de base

Les insertions Core et les modèles ORM simples sont pris en charge. Préférez les insertions Core pour les flux de données volumineux.

with engine.connect() as conn:
    conn.execute(
        events.insert(),
        [
            {"id": 13, "user": "user_1"},
            {"id": 79, "user": "user_2"},
        ],
    )

import sqlalchemy as db
from sqlalchemy import MetaData
from sqlalchemy.orm import Session, declarative_base

from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import String, UInt32
from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree

Base = declarative_base(metadata=MetaData(schema="example_db"))


class User(Base):
    __tablename__ = "users"
    __table_args__ = (MergeTree(order_by=["id"]),)

    id = db.Column(UInt32, primary_key=True)
    name = db.Column(String)


Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(id=13, name="user_1"))
    session.bulk_save_objects([User(id=79, name="user_2")])
    session.commit()

​

Migrations avec Alembic

ClickHouse Connect inclut une intégration à Alembic pour les migrations de schéma de ClickHouse. Installez-la avec :

pip install "clickhouse-connect[alembic]"

Importez clickhouse_connect.cc_sqlalchemy.alembic dans le fichier env.py d’Alembic pour enregistrer l’intégration du dialecte. L’autogénération prend en charge les évolutions courantes des tables, notamment la création et la suppression de tables, l’ajout, la modification et la suppression de colonnes, les valeurs par défaut et les commentaires. Utilisez des opérations manuelles pour renommer les tables et les colonnes. Examinez chaque migration générée avant de l’appliquer.

Les helpers op.* spécifiques à ClickHouse couvrent :

• les index de saut de données, y compris les opérations d’ajout, de matérialisation et de suppression.
• les projections, y compris les opérations d’ajout, de matérialisation et de suppression.
• la modification et la réinitialisation des paramètres de table MergeTree.
• la création et la suppression de vues matérialisées.
• la création, la suppression et le rechargement de dictionnaires.

Les index de saut de données de ClickHouse ne sont pas des index SQLAlchemy. Index, Column(index=True), op.create_index et op.drop_index sont rejetés afin d’éviter un DDL partiel ou incorrect. Utilisez op.add_clickhouse_index et op.drop_clickhouse_index.

Consultez l’exemple complet d’Alembic. Les utilisateurs qui migrent depuis clickhouse-sqlalchemy devraient également lire le guide de migration.

​

Portée et limites

• ClickHouse ne fournit pas de transactions traditionnelles via ce dialecte HTTP. engine.begin() et Session.commit() organisent le travail côté Python, mais commit et rollback sont des opérations sans effet côté serveur.
• UPDATE, les transactions en deux phases, les séquences, RETURNING et les niveaux d’isolation avancés ne sont pas implémentés par ce dialecte. Utilisez du ClickHouse SQL explicite pour les mutations côté serveur si nécessaire.
• Column(..., primary_key=True) fournit l’identité de l’objet SQLAlchemy. Cela ne crée pas de contrainte d’unicité côté serveur. Définissez le tri et, si nécessaire, les expressions de clé primaire via le moteur de table.
• Les métadonnées traditionnelles de clés étrangères, de contraintes d’unicité et d’index standard ne sont pas disponibles, car ClickHouse n’applique pas ces contraintes.
• La gestion des relations ORM, les mises à jour de type unit-of-work, les cascades, ainsi que le chargement eager ou lazy des relations, ne font pas partie du périmètre ORM pris en charge.