SQLAlchemy support

ClickHouse SQLAlchemy and Alembic support

ClickHouse Connect includes the clickhousedb SQLAlchemy dialect on top of the core driver. The synchronous dialect supports SQLAlchemy 1.4.40 and later, including SQLAlchemy 2.x, with a focus on Core queries, ClickHouse DDL, reflection, and simple ORM inserts. The async dialect requires SQLAlchemy 2.0.44 or later.

Install the SQLAlchemy dependencies with the package extra:

pip install "clickhouse-connect[sqlalchemy]"

Connect with SQLAlchemy

Create an engine with either the clickhousedb:// or clickhousedb+connect:// URL form:

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)

ClickHouse session IDs

Each pooled connection in either the synchronous or async dialect generates a distinct ClickHouse session ID by default. When requests for that connection reach the same ClickHouse server process, settings changed with SET and temporary tables persist for that connection. Named-session state and same-session overlap checks are process-local. On one server process, an overlapping request for the same user and session ID is rejected immediately with server code 373 instead of being queued. If you configure a fixed session_id, use pool_size=1, max_overflow=0 or serialize access before requests reach ClickHouse. In ClickHouse Cloud or other load-balanced deployments, requests with the same session ID can reach different servers, so do not rely on a fixed session_id as distributed state or as a distributed mutex.

Async connections

The async dialect requires SQLAlchemy 2.0.44 or later and uses the native ClickHouse Connect AsyncClient. Install its dependencies and create an async engine with the clickhousedb+async:// URL:

pip install "clickhouse-connect[sqlalchemy-async]"

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine


async def main():
    engine = create_async_engine(
        "clickhousedb+async://user:password@host:8123/mydb"
    )
    try:
        async with engine.connect() as conn:
            version = (await conn.execute(text("SELECT version()"))).scalar_one()
            print(version)
    finally:
        await engine.dispose()


asyncio.run(main())

Results are buffered. Server-side cursors are disabled, so AsyncConnection.stream() raises InvalidRequestError. AsyncSession.stream() is accepted by SQLAlchemy, but the dialect buffers the full result before returning it. Use the native AsyncClient streaming methods for large results. The raw native client is available as driver_connection while its SQLAlchemy connection is checked out:

async def stream_events(engine):
    async with engine.connect() as conn:
        raw_connection = await conn.get_raw_connection()
        client = raw_connection.driver_connection
        async with await client.query_rows_stream("SELECT * FROM events") as rows:
            async for row in rows:
                print(row)

Do not use the SQLAlchemy connection concurrently with its raw client. Finish raw-client streams before leaving the SQLAlchemy connection block, and do not keep the raw client after the connection returns to the pool. SQLAlchemy owns the borrowed client’s lifecycle, so never call client.close() or any of its private lifecycle methods. SQLAlchemy’s pool owns connection concurrency. Each pooled connection owns one native async client and defaults its aiohttp connector limits to one connection and one connection per host. Set connector_limit, connector_limit_per_host, or keepalive_timeout in the URL or connect_args to override those transport settings. With pool_pre_ping=True, SQLAlchemy checks reused connections with SELECT 1 when a pooled connection is checked out.

Async SQLAlchemy executemany inserts currently send one HTTP request for each parameter set instead of using the driver’s Native bulk insert protocol. Use this path only for small batches. For bulk data, use the pool-owned driver_connection access pattern above and await client.insert() before returning the SQLAlchemy connection to the pool. Because async executemany uses query parameter binding, naive datetime values follow naive_datetime_binding, not the naive_datetime_insert setting used by synchronous Native executemany. Typed SQLAlchemy DateTime64 binds preserve fractional seconds with both client-side and server-side parameters. Untyped %s or %(name)s parameters passed to exec_driver_sql() retain the default whole-second formatting for naive datetime values. Use timezone-aware values for unambiguous timezone behavior. Use client.insert() for Native bulk semantics.

Create and dispose an async engine in the event loop where it is used. Return every checked-out connection, then await engine.dispose() during shutdown and before using the engine from another event loop. If the engine’s owning loop has already closed, await engine.dispose() in the current loop before reuse. aiohttp may still report an unclosed transport when cleanup begins only after the owning loop has closed, so dispose before transfer when possible. pool_pre_ping=True is not a replacement for disposal when moving a pooled async engine between event loops. To share one engine across event loops without retaining loop-bound connections, configure poolclass=NullPool. If disposal runs while a connection is still checked out, the dialect closes that connection when it is returned or garbage collected. Do not call engine.sync_engine.dispose() from synchronous code. SQLAlchemy cannot await async connection cleanup there and may log the error instead of closing pooled transports.

URL query parameters can contain ClickHouse settings, ClickHouse Connect client options such as compression, query_limit, and timeouts, or HTTP/TLS options such as ca_cert. Prefix a ClickHouse setting with ch_ to force it to be treated as a server setting when needed, for example ch_http_max_field_name_size=99999.

See Connection arguments and settings for the available client options.

Run synchronous SQLAlchemy helpers such as DDL and inspection through AsyncConnection.run_sync():

from sqlalchemy import inspect


async def prepare_schema(engine, metadata):
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)
        return await conn.run_sync(
            lambda sync_conn: inspect(sync_conn).get_table_names()
        )

Per-query settings

Pass ClickHouse settings through SQLAlchemy execution options. Settings can be set on an engine, connection, or statement. A statement value takes precedence over a connection or engine value with the same key.

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()

Per-query read formats

Set ClickHouse read formats on an engine, connection, or statement through SQLAlchemy execution options with query_formats, with statement formats applied first so they override matching connection or engine keys and wildcards.

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()

Error handling

Errors raised by the driver through a SQLAlchemy connection use the DB-API classes exported from clickhouse_connect.dbapi. They are the same class objects as the corresponding classes in clickhouse_connect.driver.exceptions, so SQLAlchemy wraps them in the matching sqlalchemy.exc.DBAPIError subclass. StreamFailureError is an OperationalError and is wrapped as sqlalchemy.exc.OperationalError.

If caller cancellation can interrupt an explicit AsyncConnection.invalidate(), run invalidation in an owned task and wait for that task before propagating cancellation. This lets SQLAlchemy finish its connection-record bookkeeping:

import asyncio


async def invalidate_safely(connection):
    invalidate_task = asyncio.create_task(connection.invalidate())
    cancellation = None
    while not invalidate_task.done():
        try:
            await asyncio.wait({invalidate_task})
        except asyncio.CancelledError as ex:
            cancellation = ex
    if cancellation is not None:
        try:
            invalidate_task.result()
        finally:
            raise cancellation
    invalidate_task.result()

Do not use the connection while its invalidation task is still running. If a direct await connection.invalidate() is cancelled and connection.invalidated remains false, await connection.invalidate() again to finish cleanup before using or closing the connection.

Server-side parameters

SQLAlchemy normally renders client-side parameters. Opt in to ClickHouse server-side parameters when creating the engine:

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

Use the same server_side_params=True argument with create_async_engine() for the async dialect.

In this mode every bound value must have a ClickHouse-compatible SQLAlchemy type. Supported IN lists become typed ClickHouse Array parameters. The compiler raises CompileError when it cannot derive a compatible type or safely process a bind.

Bind names must be ClickHouse ASCII BareWord names. Names that start and end with $ are rejected because the core driver reserves them for raw binary query parameters.

Core queries

The dialect supports SQLAlchemy Core SELECT queries with joins, filters, ordering, limits and offsets, DISTINCT, and compound selects.

SQLAlchemy union(), intersect(), and except_() compile to ClickHouse UNION DISTINCT, INTERSECT DISTINCT, and EXCEPT DISTINCT. Their union_all(), intersect_all(), and except_all() counterparts compile to the corresponding ALL operators. This explicit mapping preserves SQLAlchemy’s duplicate semantics regardless of ClickHouse set-operation defaults.

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()

Lightweight DELETE is supported and requires an explicit WHERE clause:

from sqlalchemy import delete

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

Literal rendering

When SQLAlchemy inlines a bound value through literal_binds or literal_execute, the dialect uses ClickHouse quoting for generic string types and ClickHouse types. This also applies through TypeDecorator wrappers and with_variant() selections. String values retain percent signs and backslashes even when other bound parameters remain.

Python datetime values with a ClickHouse DateTime64 SQLAlchemy type retain their microseconds in client-side parameters and inline literals, including nullable values and values nested in arrays and tuples. ClickHouse applies the declared precision. Python datetime provides up to six fractional digits. Plain DateTime values retain whole-second formatting. For a text() statement, supply the type explicitly with bindparam("ts", type_=DateTime64(6)) to preserve fractional seconds.

SQLAlchemy column types must match the server schema. Declaring DateTime64 over a server DateTime column renders fractional seconds and can raise conversion errors on insert and in IN comparisons.

On SQLAlchemy 2.x, inline literals of generic sqlalchemy.ARRAY types containing ClickHouse Tuple items need dimensions=1, or the appropriate higher dimension count for nested arrays, so SQLAlchemy treats each tuple as one item. SQLAlchemy 1.4 does not support inline literals for generic ARRAY types.

If a named datetime parameter is reused, every occurrence needs a compatible DateTime64 bind type to preserve fractions. An untyped occurrence or a conflicting type keeps whole-second formatting. Set type_=DateTime64(6) on each bindparam, or use distinct parameter names with the appropriate types.

JSON type hints

Declare typed JSON paths with the typed_paths mapping. A path type can be a ClickHouse SQLAlchemy type class, a configured instance, or a ClickHouse type name string. Type name strings support types without a SQLAlchemy constructor, such as Dynamic, and can still be used for complex configured type expressions. They preserve names in a named Tuple.

Type name strings can contain configured nested JSON types such as Array(JSON(`child` UInt32)). Recognized ClickHouse type names are case-insensitive in these strings and are emitted with their canonical capitalization. A string must contain one complete type expression. Trailing text and malformed nested JSON arguments are rejected.

An empty Tuple() is not supported as a JSON typed path because ClickHouse cannot serialize it through a JSON column’s Native format. The core driver supports Tuple() in query and insert columns at any position, including nested in positional or named tuples, inside Array, and as Nullable(Tuple()) where enabled by the server.

from sqlalchemy import Column, MetaData, Table

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

events = Table(
    "events",
    MetaData(),
    Column(
        "payload",
        JSON(
            typed_paths={
                "event.id": UInt32,
                "details": "Tuple(id UInt32, label Nullable(String))",
                "attributes": "Variant(String, Array(String))",
            },
            max_dynamic_paths=256,
            max_dynamic_types=16,
            skip_paths=["internal.debug"],
            skip_regexps=[r"^private\."],
        ),
    ),
)

For simple Python identifier paths, keyword arguments are shorthand for typed_paths, for example JSON(user_id=UInt32). Use typed_paths for dotted paths, spaces, backticks, %2E encoded dots, or names that match constructor options. A typed path named SKIP is supported through the mapping. Keys in typed_paths and values in skip_paths are decoded names. Leading or trailing backticks and double quotes are treated as literal path characters, not as pre-applied SQL quoting. Inside a raw type string, backticks and double quotes are ClickHouse identifier syntax.

Up to 1000 typed paths can be configured. max_dynamic_paths accepts 0 through 10000. max_dynamic_types accepts 0 through 254. These ranges also apply inside raw nested JSON type strings. Explicit server defaults of 1024 and 32 are omitted from generated DDL. Plain skip paths are deduplicated. Regular expression strings are not validated by Python because ClickHouse uses RE2 syntax. Duplicate regular expressions are preserved.

A plain skip path cannot be named exactly REGEXP because ClickHouse reserves that token for SKIP REGEXP. Names such as REGEXP_foo remain valid. In a raw JSON type string, a plain SKIP operand must be one ClickHouse identifier or a dot-separated compound identifier. An unquoted compound identifier cannot start with REGEXP; quote that first component when it is path data. SKIP REGEXP must have one single-quoted string literal. Quote identifier parts with backticks or double quotes when they contain spaces or punctuation. Raw JSON type hints support Variant(...); standalone Variant has no public SQLAlchemy constructor. Variant members are ordered and deduplicated by the same canonical names used by ClickHouse.

The constructor orders arguments in the same canonical form returned by ClickHouse. Reflected types, SQLAlchemy type copies, and Alembic autogeneration preserve the configuration.

JSON subcolumns

For a column declared or reflected as ClickHouse JSON, use square brackets to select one segment of a storage-backed subcolumn path at a time:

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"] compiles to ClickHouse dotted identifier syntax. Each part is quoted separately, for example `events`.`payload`.`severity`. It reads ClickHouse’s stored JSON subcolumn and does not call getSubcolumn. Chain [] or .subcolumn() once for each path segment. Each segment must be a non-empty string.

Passing type_ to .subcolumn() wraps the dotted path in a SQL CAST and assigns that type to the SQLAlchemy expression. Without type_, .subcolumn("segment") behaves like ["segment"].

An untyped path has ClickHouse’s Dynamic type. ClickHouse does not allow Dynamic values directly in ORDER BY or GROUP BY. Pass type_ when a subcolumn is used there.

For statically typed code, import json_subcolumn from clickhouse_connect.cc_sqlalchemy. The helper also takes one segment at a time and preserves the Python result type from 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)

In this example, type checkers see request_id as ColumnElement[int].

Each segment is quoted independently, including names with spaces or backticks. Backticks do not make a dot literal to ClickHouse JSON path handling. When json_type_escape_dots_in_keys is enabled, use ClickHouse’s %2E encoding for literal dots in keys. Access a key named a.b as payload["a%2Eb"], not payload["a.b"].

ClickHouse query extensions

Import select from clickhouse_connect.cc_sqlalchemy to expose typed ClickHouse methods to static type checkers. The standard sqlalchemy.select also has these methods at runtime.

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)
)

The ClickHouse Select methods are:

Method	SQL feature
.final()	FINAL for a table
.sample(value)	SAMPLE, using a fraction, row count, or expression
.prewhere(expression)	PREWHERE; repeated calls combine with AND
.limit_by(columns, limit, offset=None)	LIMIT ... BY
.array_join(...)	ARRAY JOIN
.left_array_join(...)	LEFT ARRAY JOIN
.ch_join(...)	ClickHouse joins with strictness, distribution, using, and cross options
.cte(name, materialized=True)	WITH name AS MATERIALIZED (...)

SQLAlchemy’s Select.with_hint() is a table hint API. The ClickHouse dialect does not render table hints. An applicable wildcard or clickhousedb hint emits an SAWarning and leaves the generated SQL unchanged. Use final(), sample(), prewhere(), or limit_by() for those ClickHouse clauses.

Select.with_statement_hint() is a raw tail directive API. It appends the supplied text to the end of the SELECT without ClickHouse-specific validation. This remains available for trusted static SQL such as SETTINGS max_threads=1:

stmt = select(events.c.id).with_statement_hint("SETTINGS max_threads=1")

For ClickHouse settings, prefer execution options so the driver handles the settings separately from the SQL text:

stmt = select(events.c.id).execution_options(settings={"max_threads": 1})

For example, a ClickHouse GLOBAL ANY LEFT JOIN can be chained without nesting a custom FromClause:

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",
    )
)

Use the explicit Lambda construct for ClickHouse higher-order functions:

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")
)

The standard SQLAlchemy values() construct compiles to ClickHouse’s VALUES table-function syntax, including when used in a common table expression. The CTE form requires SQLAlchemy 2.0.42 or later, where Values.cte() was added.

Materialized CTEs

By default ClickHouse inlines a common table expression, so a CTE referenced more than once has its body executed once per reference. Pass materialized=True to .cte() to emit WITH <name> AS MATERIALIZED (...), which computes the body once:

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})
)

The server materializes the CTE only when the keyword is present, enable_materialized_cte=1, and the analyzer is enabled. Set enable_materialized_cte on the statement, connection, or engine as shown in Per-query settings. The analyzer is enabled by default on every server that supports this feature, so setting enable_analyzer=1 explicitly is defensive. enable_materialized_cte is an experimental ClickHouse setting. With enable_materialized_cte=0 or enable_analyzer=0, the query succeeds and returns the same rows. ClickHouse silently ignores MATERIALIZED and inlines the CTE again, so a forgotten setting costs performance without raising anything. Materialized CTEs require ClickHouse 26.3 or later. Older servers reject the keyword as a syntax error.

For a statement built with the standard sqlalchemy.select, use the module-level cte() instead. It takes the statement as its first argument and otherwise mirrors 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)

The keyword renders only on the ClickHouse dialect, so a statement shared with another backend compiles unchanged there.

ClickHouse does not support recursive materialized CTEs. The SQLAlchemy helpers raise ValueError when recursive=True and materialized=True are both set.

DDL and reflection

ClickHouse Connect provides ClickHouse data types, table engines, dictionary constructs, database DDL, and table reflection.

Standalone Variant columns reflect through an internal SQLAlchemy type, and Alembic autogenerate preserves their canonical raw type names without repeated type changes. Geometry and MultiPoint columns reflect as public SQLAlchemy types.

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

Reflected columns carry server_default for DEFAULT expressions and dialect-specific attributes such as clickhouse_codec, clickhouse_ttl, clickhouse_materialized, and clickhouse_alias when present.

String values in DEFAULT, MATERIALIZED, ALIAS, and TTL clauses use ClickHouse string escaping. The same escaping applies to table, dictionary, and column comments, including comments emitted by Alembic.

MergeTree key arguments such as order_by, partition_by, primary_key, sample_by, and ttl accept SQLAlchemy column and SQL expressions as well as plain strings.

Memory(), Log(), StripeLog(), TinyLog(), Null(), and Set() accept zero arguments and round-trip through Alembic autogeneration. The existing dictionary argument remains supported. Use settings={...} to supply engine settings.

SummingMergeTree and ReplicatedSummingMergeTree accept an optional, keyword-only columns argument. Existing positional arguments keep their meaning, so SummingMergeTree("id") still sets ORDER BY id.

from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import SummingMergeTree

engine_clause = SummingMergeTree("id", columns=("delta", "n_tx"))
# Sum delta and n_tx for rows with the same id.

Pass a string, a SQLAlchemy column, a mapped column attribute, or a non-empty list or tuple of those values. List and tuple string items are quoted as identifiers. A scalar string supplies raw SQL, such as "delta" or "(delta, n_tx)". The server requires identifiers for these columns. Omit columns to let ClickHouse select the columns to sum. Reflection and Alembic autogeneration preserve an explicit column list.

Inserts and basic ORM use

Core inserts and simple ORM models are supported. For the synchronous dialect, prefer Core executemany inserts for compatible bulk data paths. For async bulk inserts, use the native AsyncClient.insert() path described in Async connections.

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

For the synchronous dialect, plain Core executemany inserts generated by the SQLAlchemy compiler use one Native bulk insert. Async executemany sends one request per parameter set, as described in Async connections. Raw SQL and inserts with expressions or other semantics that cannot be routed safely preserve the original SQL and execute once for each parameter set. If a later parameter set fails, rows written by earlier parameter sets remain committed.

Explicit multi-row insert(events).values([...]) statements work with dictionary rows, tuples in table column order, and per-row SQL expressions. Pandas to_sql(method="multi") uses this form. It inserts the rows but returns 0 because textual INSERT statements report a row count of 0 through the DB-API cursor. SQLAlchemy determines the column list from the first row. Extra dictionary keys in later rows and tuple values outside that selected column list are ignored. A later row missing a selected value fails compilation. Give every row the same columns.

With the default HTTP form limits in ClickHouse 26.4 and newer, server_side_params=True is suitable only for small explicit batches, below about 1000 bind values with headroom for other fields. Server configuration can raise this ceiling. For large plain batches with the synchronous dialect, pass rows as the second argument to execute() so the driver can use its Native bulk insert path. For async bulk data, await the native AsyncClient.insert() method.

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()

Alembic migrations

ClickHouse Connect includes Alembic integration for ClickHouse schema migrations. Install it with:

pip install "clickhouse-connect[alembic]"

For migrations through the async dialect, install both extras:

pip install "clickhouse-connect[alembic,sqlalchemy-async]"

Create an async Alembic project, then replace its generated environment with the ClickHouse-aware example:

alembic init -t async alembic

The generated alembic.ini uses script_location = %(here)s/alembic. Keep that setting when the migration directory is named alembic, or update it to the directory passed to alembic init. Replace alembic/env.py with the checked-in async Alembic env.py example, then set sqlalchemy.url in alembic.ini.

Import clickhouse_connect.cc_sqlalchemy.alembic in Alembic’s env.py to register the dialect integration. Autogenerate supports common table evolution, including table creation and removal, column add/alter/drop, defaults, and comments. Use manual operations for table and column renames. Review every generated migration before applying it.

Alembic’s migration functions remain synchronous. An async environment creates an AsyncEngine, opens an AsyncConnection, and passes the synchronous migration function to await connection.run_sync(...). Offline migrations call context.configure(url=..., literal_binds=True, dialect_opts={"paramstyle": "named"}) directly and do not create an engine. The checked-in async Alembic env.py example includes both paths and reads the connection URL through Alembic’s standard sqlalchemy.url configuration. It keeps the ClickHouse Alembic hooks and options from the worked example, including include_object, make_include_name(...), clickhouse_writer, and version_table. Do not use engine.sync_engine to run or dispose async migrations.

ClickHouse-specific op.* helpers cover:

• Data skipping indexes, including add, materialize, and drop operations.
• Projections, including add, materialize, and drop operations.
• MergeTree table setting modification and reset.
• Materialized view creation and removal.
• Dictionary creation, removal, and reload.

ClickHouse data skipping indexes are not SQLAlchemy indexes. Index, Column(index=True), op.create_index, and op.drop_index are rejected to avoid partial or incorrect DDL. Use op.add_clickhouse_index and op.drop_clickhouse_index.

See the complete Alembic worked example. Users migrating from clickhouse-sqlalchemy should also read the migration guide.

Scope and limitations

• ClickHouse does not provide traditional transactions through this HTTP dialect. engine.begin() and Session.commit() organize Python-side work, but commit and rollback are no-ops on the server.
• UPDATE, two-phase transactions, sequences, RETURNING, and advanced isolation levels are not implemented by the dialect. Use explicit ClickHouse SQL for server mutations when needed.
• Column(..., primary_key=True) supplies SQLAlchemy object identity. It does not create a server-side uniqueness constraint. Define sorting and optional primary-key expressions through the table engine.
• Traditional foreign-key, unique-constraint, and standard index metadata are not available because ClickHouse does not enforce those constraints.
• ORM relationship management, unit-of-work updates, cascades, and eager or lazy relationship loading are outside the supported ORM scope.