ClickHouse support

ClickHouse support (behind the [clickhouse] extra).

class alchemiq.clickhouse.AggregatingMergeTree(order_by, partition_by=None, primary_key=None, ttl=None, sample_by=None, settings=None)[source]

Bases: _Engine

AggregatingMergeTree engine - merges rows by aggregating AggregateFunction columns.

Parameters:
class alchemiq.clickhouse.BufferedInserter(repo, *, max_rows=10000, flush_interval=5.0, max_buffered=None)[source]

Bases: object

Accumulates rows in memory; flushes on max_rows or every flush_interval seconds.

Obtain via ClickHouseRepository.buffered() rather than constructing directly. Use as an async context manager to ensure all buffered rows are flushed on exit.

On flush failure the batch is retained and retried on the next flush call.

See also

ClickHouseRepository.buffered() - factory method.

Parameters:
async add(obj)[source]

Buffer obj and flush immediately if max_rows is reached.

Parameters:

obj (Any) – Model instance to buffer.

Raises:

ClickHouseError – if max_buffered is set and the buffer is already full.

Return type:

None

async add_many(objs)[source]

Buffer each item in objs, flushing on max_rows after each addition.

Parameters:

objs (Any)

Return type:

None

async flush()[source]

Drain the in-memory buffer and bulk-insert any pending rows immediately.

Return type:

None

async close()[source]

Stop the background timer and flush any remaining buffered rows.

Return type:

None

class alchemiq.clickhouse.ClickHouseModel(**kwargs)[source]

Bases: DeclarativeBase

Annotation-first base for ClickHouse tables (separate metadata/registry).

Declare fields like a normal alchemiq Model and set the table engine in Meta. Every concrete subclass must provide Meta.engine (MergeTree, ReplacingMergeTree, or AggregatingMergeTree); omitting it raises ConfigError at class-definition time. The engine in turn requires an order_by argument (a TypeError is raised without it) - ClickHouse has no implicit primary key, and the ORDER BY key is what the MergeTree family sorts and (for FINAL / soft-delete) collapses on.

For soft-delete support, set Meta.soft_delete = True and use ReplacingMergeTree - alchemiq will inject is_deleted, _version, and deleted_at columns automatically.

E.g.:

import datetime as dt
from alchemiq.clickhouse import ClickHouseModel, MergeTree, ReplacingMergeTree
from alchemiq.clickhouse.types import DateTime64, UInt32

class PageView(ClickHouseModel):
    event_time: dt.datetime = DateTime64(3)
    user_id: int = UInt32()
    class Meta:
        engine = MergeTree(order_by=("event_time", "user_id"))

class Document(ClickHouseModel):
    key: int = UInt32()
    body: str
    class Meta:
        soft_delete = True
        engine = ReplacingMergeTree(order_by=("key",))

Note

ch_metadata and ch_mapper_registry are kept separate from the PostgreSQL registry so CH and PG models never share the same MetaData.

See also

configure_clickhouse() - connect the process-global client.

See also

ClickHouseRepository - read/write rows for this model.

Parameters:

kwargs (Any)

registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

class alchemiq.clickhouse.ClickHousePublisher(target_model, *, mapper=None)[source]

Bases: object

Outbox Publisher adapter that batch-inserts messages into a ClickHouse model.

Used as the publisher argument to Relay when the outbox relay target is a ClickHouse table rather than a message broker.

E.g.:

from alchemiq.clickhouse.publisher import ClickHousePublisher
from alchemiq.outbox.relay import Relay

publisher = ClickHousePublisher(EventLog)
relay = Relay(publisher)
Parameters:

  • target_model (type) – A ClickHouseModel subclass that receives the inserted rows.

  • mapper (Callable[[OutboxMessage], dict[str, Any]] | None) – Optional callable (OutboxMessage) -> dict to customise how each message is mapped to a row. When None, the default mapping copies the standard outbox fields (id, topic, payload, headers, aggregate_type, aggregate_id, event_type) that exist on the model.

async publish(message)[source]

Insert a single outbox message into the target CH model.

Parameters:

message (OutboxMessage)

Return type:

None

async publish_batch(messages)[source]

Map messages to target-model rows and bulk-insert them in one CH call.

Parameters:

messages (list[OutboxMessage])

Return type:

None

class alchemiq.clickhouse.ClickHouseRepository(model=None)[source]

Bases: Generic[M]

Data-access surface for one ClickHouse model.

No transactions, UnitOfWork, signals, or cache - ClickHouse is append-only. Instantiate directly with a model class, or subclass with a type parameter:

E.g.:

from alchemiq.clickhouse import ClickHouseRepository

# direct:
repo = ClickHouseRepository(PageView)
await repo.insert(PageView(event_time=now, user_id=42))

# typed subclass:
class PageViewRepo(ClickHouseRepository[PageView]):
    pass

repo = PageViewRepo()
rows = await repo.filter(user_id=42).order_by("event_time").all()

See also

ClickHouseModel - declare the table schema.

See also

ClickHouseQuerySet - chainable query builder returned by filter, order_by, and related methods.

Parameters:

model (type[M])

filter(*args, **lookups)[source]

Return a queryset for this model with an additional WHERE condition.

Parameters:

  • args (Q)

  • lookups (Any)

Return type:

ClickHouseQuerySet

exclude(*args, **lookups)[source]

Return a queryset for this model with an additional negated WHERE condition.

Parameters:

  • args (Q)

  • lookups (Any)

Return type:

ClickHouseQuerySet

order_by(*fields)[source]

Return a queryset ordered by fields (prefix ‘-’ to reverse).

Parameters:

fields (str)

Return type:

ClickHouseQuerySet

limit(n)[source]

Return a queryset with a LIMIT clause.

Parameters:

n (int)

Return type:

ClickHouseQuerySet

offset(n)[source]

Return a queryset with an OFFSET clause.

Parameters:

n (int)

Return type:

ClickHouseQuerySet

distinct()[source]

Return a queryset with SELECT DISTINCT.

Return type:

ClickHouseQuerySet

only(*fields)[source]

Return a queryset that selects only fields (column projection).

Parameters:

fields (str)

Return type:

ClickHouseQuerySet

final()[source]

Return a queryset with FINAL appended to the FROM clause.

Return type:

ClickHouseQuerySet

with_deleted()[source]

Return a queryset that includes soft-deleted rows (raw history, no FINAL).

Return type:

ClickHouseQuerySet

only_deleted()[source]

Return a queryset that returns only soft-deleted rows (no FINAL).

Return type:

ClickHouseQuerySet

async all()[source]

Fetch all rows and return them as model instances.

Return type:

list[M]

async first()[source]

Return the first row as a model instance, or None if the table is empty.

Return type:

M | None

async count(*args, **lookups)[source]

Return the number of rows matching the optional filters.

Parameters:

  • args (Q)

  • lookups (Any)

Return type:

int

async exists(*args, **lookups)[source]

Return True if at least one row matches the optional filters.

Parameters:

  • args (Q)

  • lookups (Any)

Return type:

bool

async get_or_none(*args, **lookups)[source]

Return exactly one matching instance, None if none match.

Parameters:

  • args (Q) – Q expressions to filter on.

  • lookups (Any) – Keyword filter arguments (column=value).

Returns:

The matching instance, or None if no rows match.

Raises:

MultipleResultsFound – if more than one row matches.

Return type:

M | None

iterate(batch_size)[source]

Stream all rows in batches of batch_size via clickhouse-connect block streaming.

Parameters:

batch_size (int)

Return type:

AsyncIterator[list[M]]

async raw(sql, params=None, *, as_model=False)[source]

Execute a raw SQL string and return results.

E.g.:

repo = ClickHouseRepository(Sale)
rows = await repo.raw(
    "SELECT region, sum(amount) AS total FROM _sale GROUP BY region ORDER BY region"
)
# rows == [{"region": "EU", "total": 30}, ...]
Parameters:

  • sql (str) – A literal ClickHouse SQL query string.

  • params (dict[str, Any] | None) – Optional query parameters passed to clickhouse-connect.

  • as_model (bool) – If True, hydrate rows into model instances; otherwise return plain dicts keyed by column name.

Returns:

List of dicts (default) or model instances when as_model=True.

Return type:

list[Any]

buffered(*, max_rows=10000, flush_interval=5.0, max_buffered=None)[source]

Return a BufferedInserter backed by this repository.

The inserter accumulates rows in memory and flushes them to ClickHouse when max_rows is reached or every flush_interval seconds. Use as an async context manager to guarantee the final flush on exit.

E.g.:

async with repo.buffered(max_rows=1000, flush_interval=5.0) as buf:
    for event in events:
        await buf.add(event)
# all rows flushed when the block exits
Parameters:

  • max_rows (int) – Flush automatically once this many rows accumulate (default 10 000).

  • flush_interval (float) – Maximum seconds between automatic timer-driven flushes (default 5).

  • max_buffered (int | None) – Hard cap on in-memory rows; BufferedInserter.add() raises ClickHouseError when exceeded. None (default) means unbounded.

Returns:

A new BufferedInserter instance.

Return type:

BufferedInserter

async insert(obj)[source]

Insert a single model instance into ClickHouse and return it.

E.g.:

repo = ClickHouseRepository(PageView)
view = await repo.insert(PageView(event_time=now, user_id=42))
Parameters:

obj (M) – The model instance to insert.

Returns:

The same instance (unchanged).

Return type:

M

async bulk_insert(objs)[source]

Insert multiple model instances in a single ClickHouse call and return them.

E.g.:

repo = ClickHouseRepository(PageView)
await repo.bulk_insert([
    PageView(event_time=t1, user_id=1),
    PageView(event_time=t2, user_id=2),
])
Parameters:

objs (Iterable[M]) – Iterable of model instances to insert.

Returns:

The inserted instances as a list.

Return type:

list[M]

async update(*args, **kw)[source]

Raise UnsupportedOperationError - ClickHouse does not support row UPDATE.

Parameters:
Return type:

Any

async bulk_update(*args, **kw)[source]

Raise UnsupportedOperationError - ClickHouse does not support bulk row UPDATE.

Parameters:
Return type:

Any

async get_or_create(*args, **kw)[source]

Raise UnsupportedOperationError - ClickHouse has no upsert; use insert().

Parameters:
Return type:

Any

async update_or_create(*args, **kw)[source]

Raise UnsupportedOperationError - ClickHouse has no upsert; use bulk_insert().

Parameters:
Return type:

Any

async delete(**lookups)[source]

Soft-delete a row by inserting a tombstone with is_deleted=1.

Does NOT issue a SQL DELETE. Instead it appends a new row with the same ORDER BY key and is_deleted=1; a subsequent SELECT ... FINAL collapses the key to the latest version and hides the row because is_deleted=1.

lookups must supply every column in the table’s ORDER BY key so the tombstone lands on the same key as the live row. Missing key columns are caught eagerly before any IO.

E.g.:

repo = ClickHouseRepository(Document)
await repo.insert(Document(key=1, body="hello"))
await repo.delete(key=1)   # tombstone inserted; row hidden under FINAL
Parameters:

lookups (Any) – Keyword arguments identifying the row - must cover every ORDER BY column (e.g. key=1 for a single-column key).

Raises:

UnsupportedOperationError – if Meta.soft_delete is not True or if any ORDER BY key column is missing from lookups.

Return type:

None

See also

ClickHouseRepository.restore() - un-delete the row.

See also

ClickHouseRepository.cleanup() - physically remove tombstones.

async restore(**lookups)[source]

Un-delete a row by inserting a live marker with is_deleted=0.

The inverse of delete(): inserts a new row with the same ORDER BY key, is_deleted=0, and a fresh _version timestamp so ReplacingMergeTree FINAL collapses to this latest (live) version.

Parameters:

lookups (Any) – Keyword arguments identifying the row - must cover every ORDER BY column (same constraint as delete()).

Raises:

UnsupportedOperationError – if Meta.soft_delete is not True or if any ORDER BY key column is missing from lookups.

Return type:

None

See also

ClickHouseRepository.delete() - soft-delete the row.

async cleanup()[source]

Run OPTIMIZE TABLE FINAL CLEANUP to physically remove tombstone rows.

After delete(), ClickHouse retains both the live row and the tombstone as physical rows on disk. CLEANUP instructs the merge engine to drop all physical rows for keys where is_deleted=1 once merged.

Requires Meta.soft_delete=True and allow_experimental_replacing_merge_with_cleanup enabled on the CH server (set in Meta.engine settings or the server config).

Raises:

UnsupportedOperationError – if Meta.soft_delete is not True.

Return type:

None

Warning

CLEANUP is an experimental ClickHouse feature. Enable allow_experimental_replacing_merge_with_cleanup = 1 in the engine settings or the server configuration before calling this method.

See also

ClickHouseRepository.delete() - insert the tombstone first.

class alchemiq.clickhouse.DateTime64(precision=3, **kw)[source]

Bases: FieldType

ClickHouse DateTime64 column field with configurable sub-second precision.

When nullable=True is passed, the column type is wrapped in Nullable(DateTime64) because ClickHouse requires the explicit Nullable wrapper in DDL.

Parameters:

  • precision (int) – Sub-second precision digits (0-9; default 3 = milliseconds).

  • kw (Any)

python_type

alias of datetime

column_type()[source]

Return DateTime64(precision) or Nullable(DateTime64(precision)).

Return type:

TypeEngine

class alchemiq.clickhouse.Enum8(members, **kw)[source]

Bases: FieldType

ClickHouse Enum8 column field defined by an explicit name-to-integer mapping.

Parameters:

  • members (dict[str, int]) – Dict mapping member name to integer value, e.g. {"active": 1, "inactive": 2}.

  • kw (Any)

python_type

alias of str

column_type()[source]

Return a ch.Enum8 type built from the members mapping.

Return type:

TypeEngine

class alchemiq.clickhouse.Float32(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Float32 column field (single-precision float).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of float

class alchemiq.clickhouse.Float64(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Float64 column field (double-precision float).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of float

class alchemiq.clickhouse.Int16(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Int16 column field (signed 16-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.Int32(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Int32 column field (signed 32-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.Int64(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Int64 column field (signed 64-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.Int8(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse Int8 column field (signed 8-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.LowCardinality(inner=<class 'str'>, **kw)[source]

Bases: FieldType

ClickHouse LowCardinality wrapper - dictionary-encodes a column for low-cardinality data.

Parameters:

  • inner (type) – Python type whose default CH scalar is used as the inner type (default str).

  • kw (Any)

column_type()[source]

Return LowCardinality(<inner_ch_type>).

Return type:

TypeEngine

class alchemiq.clickhouse.MergeTree(order_by, partition_by=None, primary_key=None, ttl=None, sample_by=None, settings=None)[source]

Bases: _Engine

Standard MergeTree engine - general-purpose append-only analytics table.

E.g.:

from alchemiq.clickhouse.engines import MergeTree

engine = MergeTree(
    order_by=("event_time", "user_id"),
    partition_by="toYYYYMM(event_time)",
    ttl="event_time + INTERVAL 90 DAY",
)

See also

ReplacingMergeTree - for deduplication or soft-delete.

Parameters:
class alchemiq.clickhouse.ReplacingMergeTree(order_by, partition_by=None, primary_key=None, ttl=None, sample_by=None, settings=None, version=None, is_deleted=None)[source]

Bases: _Engine

ReplacingMergeTree engine - deduplicates rows by ORDER BY key on merge.

Required for soft-delete models (Meta.soft_delete=True). When version and is_deleted are set (as ClickHouseModel does automatically for soft-delete models), SELECT ... FINAL retains the row with the highest version value and filters out rows where is_deleted=1.

E.g.:

from alchemiq.clickhouse.engines import ReplacingMergeTree

# basic dedup by version column:
engine = ReplacingMergeTree(order_by="key", version="ver")

# with soft-delete support (set automatically by Meta.soft_delete=True):
engine = ReplacingMergeTree(
    order_by=("key",),
    version="_version",
    is_deleted="is_deleted",
)
Parameters:

  • version (str | None) – Column name used as the deduplication version (latest value wins).

  • is_deleted (str | None) – Column name that marks tombstone rows (1 = deleted). When set, SELECT ... FINAL filters out rows where is_deleted=1.

  • order_by (str | tuple[str, ...])

  • partition_by (str | tuple[str, ...] | None)

  • primary_key (str | tuple[str, ...] | None)

  • ttl (str | None)

  • sample_by (str | None)

  • settings (dict[str, Any] | None)

See also

MergeTree - for non-deduplicated append-only tables.

class alchemiq.clickhouse.UInt16(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse UInt16 column field (unsigned 16-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.UInt32(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse UInt32 column field (unsigned 32-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.UInt64(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse UInt64 column field (unsigned 64-bit integer).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

class alchemiq.clickhouse.UInt8(*, python_type=None, unique=False, index=False, primary_key=False, nullable=False, default=<MISSING>, server_default=None, max_length=None, onupdate=None)[source]

Bases: _CHField

ClickHouse UInt8 column field (unsigned 8-bit integer, range 0-255).

Parameters:

  • python_type (type)

  • unique (bool)

  • index (bool)

  • primary_key (bool)

  • nullable (bool)

  • default (Any)

  • server_default (Any)

  • max_length (int | None)

  • onupdate (Any)

python_type

alias of int

alchemiq.clickhouse.configure_clickhouse(dsn=None, *, host=None, port=None, username='default', password='', database='default', secure=False, settings=None, **client_kw)[source]

Store ClickHouse connection parameters; the async client connects on first use.

Call once at application startup (e.g. inside a FastAPI/FastStream lifespan). Pass either a dsn URL or individual host/port keyword arguments. The client is created lazily on the first internal client-creation call and is shared across the process. Call dispose_clickhouse() on shutdown.

E.g.:

from alchemiq.clickhouse import configure_clickhouse, dispose_clickhouse

# host/port form:
configure_clickhouse(host="localhost", port=8123)

# DSN form:
configure_clickhouse(dsn="clickhouse://default:@localhost/default")
Parameters:

  • dsn (str | None) – ClickHouse DSN URL (e.g. clickhouse://user:pass@host/db). Mutually optional with host/port - use one or the other.

  • host (str | None) – Server hostname.

  • port (int | None) – Server port (default 8123 for HTTP, 8443 for HTTPS).

  • username (str) – ClickHouse username (default "default").

  • password (str) – ClickHouse password (default "").

  • database (str) – Target database name (default "default").

  • secure (bool) – Use TLS/HTTPS if True (default False).

  • settings (dict[str, Any] | None) – Extra ClickHouse server settings passed to the client.

  • client_kw (Any) – Additional keyword arguments forwarded to clickhouse_connect.get_async_client.

Return type:

None

See also

dispose_clickhouse() - close and reset the client on shutdown.

async alchemiq.clickhouse.create_clickhouse_tables()[source]

Create all registered CH tables using CREATE TABLE IF NOT EXISTS.

Tables are created in ch_metadata dependency order. Safe to call on every startup - existing tables are left unchanged.

See also

drop_clickhouse_tables() - tear down all CH tables.

Return type:

None

async alchemiq.clickhouse.dispose_clickhouse()[source]

Close the process-global client and clear all connection state.

Safe to call even if the client was never created (no-op in that case). Typically called in a lifespan finally block or in test teardown.

See also

configure_clickhouse() - reconfigure after disposing.

Return type:

None

async alchemiq.clickhouse.drop_clickhouse_tables()[source]

Drop all registered CH tables using DROP TABLE IF EXISTS.

Tables are dropped in reverse ch_metadata dependency order.

See also

create_clickhouse_tables() - recreate the tables.

Return type:

None

alchemiq.clickhouse.is_clickhouse_configured()[source]

Return True if configure_clickhouse() has been called and not yet disposed.

Return type:

bool

async alchemiq.clickhouse.optimize(model, *, cleanup=False)[source]

Run OPTIMIZE TABLE FINAL [CLEANUP] on model’s table.

Parameters:

  • model (type) – A ClickHouseModel subclass whose table to optimize.

  • cleanup (bool) – If True, appends CLEANUP to instruct ReplacingMergeTree to physically remove rows where is_deleted=1 once merged. Requires allow_experimental_replacing_merge_with_cleanup enabled on the CH server.

Return type:

None

See also

ClickHouseRepository.cleanup() - higher-level soft-delete cleanup.