Clickhouse SQLAlchemy

https://github.com/xzkostyan/clickhouse-sqlalchemy
Admin
Clickhouse SQLAlchemy is a dialect for SQLAlchemy that enables Python applications to connect and...
Tokens:23,556
Snippets:165
Trust Score:7.3
License:MIT
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install clickhouse-sqlalchemy from GitHub

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/installation.md

Installs the development version of clickhouse-sqlalchemy directly from its GitHub repository. This is useful for testing the latest features or contributing to the project.

```bash
pip install git+https://github.com/xzkostyan/clickhouse-sqlalchemy@master#egg=clickhouse-sqlalchemy
```

--------------------------------

### ClickHouse SQLAlchemy Native Connection Examples

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/connection.md

Illustrates connection string examples for the Native driver in ClickHouse SQLAlchemy. It includes a simple DSN, an example with LZ4 compression and SSL certificate configuration, and a DSN for connecting to multiple hosts.

```RST
clickhouse+native://host/db
```

```RST
clickhouse+native://wronghost/default?alt_hosts=localhost:9000
```

--------------------------------

### ClickHouse SQLAlchemy Asynch Connection Example

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/connection.md

Provides a simple DSN example for establishing an asynchronous connection using the Asynch driver in ClickHouse SQLAlchemy. It notes that parameters are proxied to the asynch library.

```RST
clickhouse+asynch://host/db
```

--------------------------------

### ClickHouse SQLAlchemy HTTP Connection Examples

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/connection.md

Provides examples of connection strings for the HTTP driver in ClickHouse SQLAlchemy. It demonstrates simple DSNs, connections with user credentials and HTTPS protocol, and configurations when using Nginx as a proxy.

```RST
clickhouse+http://host/db
```

```RST
clickhouse+http://user:password@host:8443/db?protocol=https
```

```RST
clickhouse+http://user:password@host:8124/test?protocol=https
```

--------------------------------

### Define Table and Insert Data with SQLAlchemy and ClickHouse

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/quickstart.md

This snippet demonstrates defining a ClickHouse table using SQLAlchemy's declarative base and inserting data into it. It requires `sqlalchemy` and `clickhouse-sqlalchemy`.

```python
from sqlalchemy import create_engine, Column, MetaData

from clickhouse_sqlalchemy import (
    Table, make_session, get_declarative_base, types, engines
)

uri = 'clickhouse+native://localhost/default'

engine = create_engine(uri)
session = make_session(engine)
metadata = MetaData(bind=engine)

Base = get_declarative_base(metadata=metadata)

class Rate(Base):
    day = Column(types.Date, primary_key=True)
    value = Column(types.Int32)

    __table_args__ = (
        engines.Memory(),
    )

# Emits CREATE TABLE statement
Rate.__table__.create()
```

--------------------------------

### Install Test Requirements and Build Package (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Installs necessary test requirements and builds the package for local development. This script assumes Python is installed and accessible.

```bash
python testsrequire.py && python setup.py develop
```

--------------------------------

### ReplicatedMergeTree Example

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates the ReplicatedMergeTree engine for setting up replicated tables in a ClickHouse cluster. It requires specifying the ZooKeeper path for replication and the replica name. The `clickhouse_cluster` argument is used for the ON CLUSTER clause.

```python
class ReplicatedTable(Base):
    __tablename__ = 'replicated_events'
    date = Column(types.Date, primary_key=True)
    event_id = Column(types.UInt64)

    __table_args__ = (
        engines.ReplicatedMergeTree(
            '/clickhouse/tables/{shard}/replicated_events',
            '{replica}',
            partition_by=date,
            order_by=(date, event_id)
        ),
        {'clickhouse_cluster': 'my_cluster'}  # ON CLUSTER clause
    )
```

--------------------------------

### Install clickhouse-sqlalchemy from PyPI

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/installation.md

Installs the clickhouse-sqlalchemy package using pip. This is the standard method for obtaining the latest stable release.

```bash
pip install clickhouse-sqlalchemy
```

--------------------------------

### Configure MergeTree Engine with Settings (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Shows how to configure the MergeTree engine with additional keyword arguments for settings. This example passes `key='value'` to the `MergeTree` engine, which will be rendered in the `SETTINGS` clause of the `CREATE TABLE` statement.

```python
engines.MergeTree(
    partition_by=date,
    key='value'
)
```

--------------------------------

### Connection Setup for ClickHouse SQLAlchemy

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates how to establish database connections to ClickHouse using SQLAlchemy's create_engine with various URIs supporting native TCP, HTTP, and async protocols. It also shows session creation and engine reflection options.

```python
from sqlalchemy import create_engine, MetaData
from clickhouse_sqlalchemy import make_session, get_declarative_base

# Native TCP connection (recommended for performance)
native_uri = 'clickhouse+native://user:password@localhost:9000/default'
engine = create_engine(native_uri)

# HTTP connection with HTTPS
http_uri = 'clickhouse+http://user:password@localhost:8443/default?protocol=https'
http_engine = create_engine(http_uri)

# Async connection for asyncio applications
async_uri = 'clickhouse+asynch://user:password@localhost:9000/default'
async_engine = create_engine(async_uri)

# Native connection with compression and SSL
import certifi
secure_uri = (
    'clickhouse+native://user:pass@host/db?compression=lz4&'
    f'secure=True&ca_certs={certifi.where()}'
)

# Native with multiple hosts for failover
failover_uri = 'clickhouse+native://primary-host/default?alt_hosts=backup-host:9000'

# Create session with ClickHouse extensions (required for FINAL, SAMPLE, etc.)
session = make_session(engine)

# For async sessions
async_session = make_session(async_engine, is_async=True)

# HTTP with custom requests.Session for connection control
from requests import Session
custom_http_engine = create_engine(
    'clickhouse+http://localhost/test',
    connect_args={'http_session': Session()}
)

# Disable engine reflection for performance with many tables
fast_uri = 'clickhouse+native://localhost/default?engine_reflection=false'
```

--------------------------------

### SummingMergeTree Example

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Illustrates the SummingMergeTree engine, which automatically aggregates specified numeric columns. This is useful for summarizing metrics like impressions and clicks. It requires defining the columns to be summed.

```python
class SummingExample(Base):
    __tablename__ = 'summing_example'
    date = Column(types.Date, primary_key=True)
    category = Column(types.String)
    impressions = Column(types.UInt64)
    clicks = Column(types.UInt64)

    __table_args__ = (
        engines.SummingMergeTree(
            columns=(impressions, clicks),  # Columns to sum
            partition_by=func.toYYYYMM(date),
            order_by=(date, category)
        ),
    )
```

--------------------------------

### Query Inserted Data with SQLAlchemy and ClickHouse

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/quickstart.md

This snippet shows how to insert multiple records into a ClickHouse table and then query the count of specific records using SQLAlchemy. It assumes the `Rate` table and `session` object are already defined.

```python
from datetime import date, timedelta

from sqlalchemy import func

today = date.today()
rates = [
    {'day': today - timedelta(i), 'value': 200 - i}
    for i in range(100)
]
session.execute(Rate.__table__.insert(), rates)

session.query(func.count(Rate.day)) \
    .filter(Rate.day > today - timedelta(20)) \
    .scalar()
```

--------------------------------

### Run Tests Locally (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Executes tests using the pytest framework. This command assumes pytest is installed and configured.

```bash
pytest -v
```

--------------------------------

### VersionedCollapsingMergeTree Example

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates the use of VersionedCollapsingMergeTree for handling versioned data, suitable for scenarios where data updates are frequent and ordering is critical. It requires specifying the sign and version columns for merging.

```python
class VersionedCollapsing(Base):
    __tablename__ = 'versioned_collapsing'
    date = Column(types.Date, primary_key=True)
    id = Column(types.UInt64)
    sign = Column(types.Int8)
    version = Column(types.UInt64)
    state = Column(types.String)

    __table_args__ = (
        engines.VersionedCollapsingMergeTree(
            sign, version,
            partition_by=date,
            order_by=(date, id)
        ),
    )
```

--------------------------------

### ClickHouse SQLAlchemy Native Connection with Compression and SSL (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/connection.md

Demonstrates a Python example for configuring a Native connection in ClickHouse SQLAlchemy, including LZ4 compression and securing the connection with a Let's Encrypt certificate.

```python
import certify

dsn = (
    'clickhouse+native://user:pass@host/db?compression=lz4&'
    'secure=True&ca_certs={}'.format(certify.where())
)
```

--------------------------------

### Using select() Function for Complex Queries with SQLAlchemy

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates how to use the `select()` function from `clickhouse_sqlalchemy` for constructing more complex queries with multiple clauses. This example combines filtering, sampling, FINAL, and LIMIT BY clauses.

```python
from datetime import date
from clickhouse_sqlalchemy import select

# Assuming session and PageViews are defined
# stmt = select(PageViews.user_id, PageViews.page) \
#     .where(PageViews.date == date.today()) \
#     .sample(0.5) \
#     .final() \
#     .limit_by([PageViews.user_id], limit=5)

# results = session.execute(stmt).fetchall()
```

--------------------------------

### Add Host Entry for ClickHouse Server (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Appends an entry to the /etc/hosts file to map 'clickhouse-server' to the local IP address 127.0.0.1. This is necessary for the Dockerized setup to resolve the hostname correctly.

```bash
echo '127.0.0.1 clickhouse-server' | sudo tee -a /etc/hosts > /dev/null
```

--------------------------------

### AggregatingMergeTree Example

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows the AggregatingMergeTree engine, designed for storing pre-aggregated states. This engine is efficient for analytical queries where intermediate aggregation results are stored. It requires defining partitioning and ordering keys.

```python
class AggregatingExample(Base):
    __tablename__ = 'aggregating_example'
    date = Column(types.Date, primary_key=True)
    category = Column(types.String)

    __table_args__ = (
        engines.AggregatingMergeTree(
            partition_by=func.toYYYYMM(date),
            order_by=(date, category)
        ),
    )
```

--------------------------------

### Define CollapsingMergeTree Engine (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Example of defining the CollapsingMergeTree engine with its required `sign` column and specifying `partition_by` and `order_by` clauses. This is a common engine for handling collapsing data.

```python
engines.CollapsingMergeTree(
    sign,
    partition_by=date,
    order_by=(date, x)
)
```

--------------------------------

### Complex MergeTree Engine Configurations (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Provides examples of complex configurations for the MergeTree engine, including specifying primary keys, order by clauses with functions, partition by clauses using `sqlalchemy.text`, and sample by clauses. These demonstrate advanced customization options.

```python
engines.MergeTree(order_by=func.tuple_())

engines.MergeTree(
    primary_key=('device_id', 'timestamp'),
    order_by=('device_id', 'timestamp'),
    partition_by=func.toYYYYMM(timestamp)
)

engines.MergeTree(
    partition_by=text('toYYYYMM(date)'),
    order_by=('date', func.intHash32(x)),
    sample_by=func.intHash32(x)
)

engines.MergeTree(
    partition_by=date,
    order_by=(date, x),
    primary_key=(x, y),
    sample_by=func.random(),
    key='value'
)
```

--------------------------------

### Start ClickHouse Client Docker Container (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Creates a ClickHouse client container linked to the ClickHouse server container. This container is kept alive indefinitely to allow host interaction. It's linked to 'test-clickhouse-server' and named 'test-clickhouse-client'. The $VERSION variable should be replaced with the desired ClickHouse version.

```bash
docker run --rm --entrypoint "/bin/sh" --name test-clickhouse-client --link test-clickhouse-server:clickhouse-server clickhouse/clickhouse-client:$VERSION -c 'while :; do sleep 1; done'
```

--------------------------------

### Start ClickHouse Server Docker Container (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Launches a ClickHouse server instance within a Docker container. It maps ports 9000 and 8123 from the container to the host machine and names the container 'test-clickhouse-server'. The $VERSION variable should be replaced with the desired ClickHouse version.

```bash
docker run --rm -p 127.0.0.1:9000:9000 -p 127.0.0.1:8123:8123 --name test-clickhouse-server clickhouse/clickhouse-server:$VERSION
```

--------------------------------

### UNION ALL Query Example

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Illustrates how to perform a `UNION ALL` operation between two select statements using SQLAlchemy's `union_all` function. This is useful for combining results from different tables or queries.

```python
from sqlalchemy import union_all

select_rate = session.query(
    Rate.day.label('date'),
    Rate.value.label('x')
)
select_another_rate = session.query(
    another_table.c.day.label('date'),
    another_table.c.value.label('x')
)

union_all(select_rate, select_another_rate) \
    .execute() \
    .fetchone()
```

--------------------------------

### MergeTree with TTL Example

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Illustrates the MergeTree engine configured with Time-To-Live (TTL) settings for automatic data expiration and movement. It uses `ttl_delete` and `ttl_to_disk` functions to define expiration policies based on date.

```python
from clickhouse_sqlalchemy.sql.ddl import ttl_delete, ttl_to_disk, ttl_to_volume

class TTLExample(Base):
    __tablename__ = 'ttl_example'
    date = Column(types.Date, primary_key=True)
    data = Column(types.String)

    __table_args__ = (
        engines.MergeTree(
            partition_by=date,
            order_by=(date,),
            ttl=[
                ttl_delete(date + func.toIntervalDay(30)),  # Delete after 30 days
                ttl_to_disk(date + func.toIntervalDay(7), 'cold_storage'),  # Move after 7 days
            ]
        ),
    )
```

--------------------------------

### Batch INSERT Statement

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Provides an example of performing a batch INSERT operation into a ClickHouse table using SQLAlchemy. It shows how to prepare a list of dictionaries representing rows and execute them using `session.execute` with the table's `insert()` method.

```python
from datetime import date, timedelta
from sqlalchemy import func

today = date.today()
rates = [
    {'day': today - timedelta(i), 'value': 200 - i}
    for i in range(100)
]

# Emits single INSERT statement.
session.execute(table.insert(), rates)
```

--------------------------------

### ARRAY JOIN Clause in SQLAlchemy

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates the usage of the ARRAY JOIN clause in ClickHouse through SQLAlchemy. This clause is used to flatten arrays into rows. Examples are provided for both session.query and select.

```python
session.query(...).array_join(...)
```

```python
select([...]).array_join(...)
```

--------------------------------

### Configure ClickHouse Query Execution with Settings and Streaming in Python

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates how to configure query execution in ClickHouse using SQLAlchemy, including setting custom server parameters like 'max_threads' and 'max_memory_usage'. It also shows how to process large datasets efficiently using streaming with 'yield_per' and how to use external tables for filtering data.

```python
from sqlalchemy import create_engine, Column, MetaData, func
from clickhouse_sqlalchemy import Table, make_session, get_declarative_base, types, engines

engine = create_engine('clickhouse+native://localhost/default')
session = make_session(engine)
metadata = MetaData(bind=engine)
Base = get_declarative_base(metadata=metadata)

class LargeTable(Base):
    id = Column(types.UInt64, primary_key=True)
    data = Column(types.String)
    value = Column(types.Int32)
    __table_args__ = (engines.MergeTree(order_by=(id,)),)

# Query with custom ClickHouse settings
settings = {
    'max_threads': 4,              # Parallel execution threads
    'priority': 5,                 # Query priority (lower = higher priority)
    'max_memory_usage': 10000000000,  # 10GB memory limit
    'max_execution_time': 300,     # 5 minute timeout
}

results = session.query(LargeTable) \
    .filter(LargeTable.value > 100) \
    .execution_options(settings=settings) \
    .all()

# Streaming results with yield_per (native driver only)
# Fetches results in chunks to avoid loading all into memory
def process_large_result():
    for row in session.query(LargeTable).yield_per(10000):
        # Process each row - only 10000 rows in memory at a time
        process_row(row)

# External tables - pass temporary data to ClickHouse
external_data = Table(
    'user_whitelist', metadata,
    Column('user_id', types.UInt64),
    Column('name', types.String),
    clickhouse_data=[
        (1, 'Alice'),
        (2, 'Bob'),
        (3, 'Charlie'),
    ],
    extend_existing=True
)

# Query using external table for filtering
result = session.query(func.count(LargeTable.id)) \
    .filter(LargeTable.id.in_(
        session.query(external_data.c.user_id)
    )) \
    .execution_options(external_tables=[external_data]) \
    .scalar()

# Handling streaming with multiple sessions to avoid cursor conflicts
class SeparateSession:
    def __init__(self, engine):
        self.engine = engine
        self.session = None

    def __enter__(self):
        self.session = make_session(self.engine)
        return self.session

    def __exit__(self, *exc_info):
        self.session.close()

def stream_with_other_queries():
    # Create generator that streams results
    def generate_rows():
        with SeparateSession(engine) as streaming_session:
            yield from streaming_session.query(LargeTable).yield_per(5000)

    rows = generate_rows()

    # Can still run other queries while streaming
    count = session.query(func.count(LargeTable.id)).scalar()

    for row in rows:
        process_row(row)

# Apply FINAL modifier via settings (affects all tables)
final_results = session.query(LargeTable) \
    .execution_options(settings={'final': 1}) \
    .all()

def process_row(row):
    pass  # Your processing logic

```

--------------------------------

### UPDATE Statement (ALTER UPDATE)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates how SQLAlchemy's UPDATE statement is mapped to ClickHouse's `ALTER UPDATE` command. The example shows updating a column value based on a condition.

```python
tbl = Table(...)
session.execute(t1.update().where(t1.c.x == 25).values(x=5))
```

```python
tbl = Table(...)
session.execute(update(t1).where(t1.c.x == 25).values(x=5))
```

```sql
ALTER TABLE ... UPDATE x=5 WHERE x = 25
```

--------------------------------

### Create ClickHouse Client Host Script (Bash)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/development.md

Creates a bash script at /usr/local/bin/clickhouse-client that allows executing the Dockerized client from the host machine. This script forwards commands to the 'test-clickhouse-client' Docker container.

```bash
echo -e '#!/bin/bash\n\ndocker exec test-clickhouse-client clickhouse-client "$@"' | sudo tee /usr/local/bin/clickhouse-client > /dev/null
sudo chmod +x /usr/local/bin/clickhouse-client
```

--------------------------------

### DELETE Statement (ALTER DELETE)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Shows how SQLAlchemy's DELETE statement is translated into ClickHouse's `ALTER DELETE` command. The example illustrates deleting rows that meet a specific condition.

```python
tbl = Table(...)
session.execute(t1.delete().where(t1.c.x == 25))
```

```python
tbl = Table(...)
session.execute(delete(t1).where(t1.c.x == 25))
```

```sql
ALTER TABLE ... DELETE WHERE x = 25
```

--------------------------------

### Pure Alembic Migration using op.execute

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/migrations.md

Illustrates how to perform ClickHouse migrations using Alembic's `op.execute` function when autogeneration is not suitable or when using older versions of clickhouse-sqlalchemy. This allows for writing raw SQL commands.

```python
from alembic import op

# ... other migration code ...

op.execute("ALTER TABLE my_table ADD COLUMN new_column String")
```

--------------------------------

### SELECT with SAMPLE Clause

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates how to use the `SAMPLE` clause in a ClickHouse SELECT query via SQLAlchemy. This allows for sampling a fraction of the data, which can be useful for performance on large tables.

```python
session.query(table.c.x).sample(0.1)
```

```python
select([table.c.x]).sample(0.1)
```

```sql
SELECT ... FROM ... SAMPLE 0.1
```

--------------------------------

### FINAL Clause in SQLAlchemy

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Details how to use the FINAL clause with SQLAlchemy for queries involving MergeTree tables. It allows selecting the final version of rows after data merging. Examples are provided for both session.query and select.

```python
session.query(table.c.x).final().group_by(table.c.x)
```

```python
select([table.c.x]).final().group_by(table.c.x)
```

--------------------------------

### Configure Alembic Migrations for ClickHouse Tables in Python

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows how to integrate Alembic with clickhouse-sqlalchemy for database schema migrations. This includes setting up the Alembic environment and defining migration scripts to create tables with ClickHouse-specific engines and partitioning.

```python
# alembic/env.py configuration
from alembic import context
from sqlalchemy import engine_from_config, pool
from clickhouse_sqlalchemy import types, engines

# Your model's MetaData object
from myapp.models import Base
target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )

        with context.begin_transaction():
            context.run_migrations()

# Example migration file: versions/001_create_events_table.py
"""create events table

Revision ID: 001
"""
from alembic import op
import sqlalchemy as sa
from clickhouse_sqlalchemy import types, engines

revision = '001'
down_revision = None

def upgrade():
    # Create table with ClickHouse engine
    op.create_table(
        'events',
        sa.Column('event_date', types.Date, primary_key=True),
        sa.Column('event_id', types.UInt64),
        sa.Column('user_id', types.UInt64),
        sa.Column('event_type', types.String),
        sa.Column('value', types.Int32),
    )

    # Add engine separately (not auto-generated)
    op.execute("""
        ALTER TABLE events
        ENGINE = MergeTree()
        PARTITION BY toYYYYMM(event_date)
        ORDER BY (event_date, event_id)
    """)

def downgrade():
    op.drop_table('events')

```

--------------------------------

### Basic DDL Support (CREATE/DROP)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Illustrates basic Data Definition Language (DDL) support in the ClickHouse SQLAlchemy dialect, specifically for creating and dropping tables. It shows how to obtain the table object and invoke its `create()` and `drop()` methods.

```python
table = Rate.__table__
table.create()
another_table.create()

another_table.drop()
table.drop()
```

--------------------------------

### GROUP BY Modifiers (CUBE, ROLLUP, TOTALS) in SQLAlchemy

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Explains how to apply CUBE, ROLLUP, and TOTALS modifiers to GROUP BY clauses in ClickHouse using SQLAlchemy. These modifiers generate subtotals and grand totals. Examples are shown for both session.query and select.

```python
session.query(table.c.x).group_by(table.c.x).with_cube()
session.query(table.c.x).group_by(table.c.x).with_rollup()
session.query(table.c.x).group_by(table.c.x).with_totals()
```

```python
select([table.c.x]).group_by(table.c.x).with_cube()
select([table.c.x]).group_by(table.c.x).with_rollup()
select([table.c.x]).group_by(table.c.x).with_totals()
```

--------------------------------

### Configure ClickHouse Table Engines with SQLAlchemy

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows how to define various ClickHouse table engines using SQLAlchemy, including MergeTree, ReplacingMergeTree, and CollapsingMergeTree. This allows for specifying partitioning, ordering, primary keys, and other engine-specific options.

```python
from sqlalchemy import Column, create_engine, MetaData, func, text
from clickhouse_sqlalchemy import Table, get_declarative_base, types, engines

engine = create_engine('clickhouse+native://localhost/default')
metadata = MetaData(bind=engine)
Base = get_declarative_base(metadata=metadata)

# Basic MergeTree with partitioning and ordering
class BasicMergeTree(Base):
    date = Column(types.Date, primary_key=True)
    user_id = Column(types.UInt64)
    value = Column(types.Int32)

    __table_args__ = (
        engines.MergeTree(
            partition_by=func.toYYYYMM(date),
            order_by=(date, user_id),
            primary_key=(date,),  # Can differ from order_by
            sample_by=func.intHash32(user_id),  # For SAMPLE queries
        ),
    )

# ReplacingMergeTree - deduplicates by order_by key
class ReplacingExample(Base):
    __tablename__ = 'replacing_example'
    id = Column(types.UInt64, primary_key=True)
    version = Column(types.UInt64)
    data = Column(types.String)

    __table_args__ = (
        engines.ReplacingMergeTree(
            version=version,  # Keep row with highest version
            order_by=(id,)
        ),
    )

# CollapsingMergeTree - for state tracking with sign column
class CollapsingExample(Base):
    __tablename__ = 'collapsing_example'
    date = Column(types.Date, primary_key=True)
    user_id = Column(types.UInt64)
    sign = Column(types.Int8)  # 1 for insert, -1 for delete
    value = Column(types.Int32)

    __table_args__ = (
        engines.CollapsingMergeTree(
            sign,
            partition_by=func.toYYYYMM(date),
            order_by=(date, user_id)
        ),
    )

```

--------------------------------

### Define Tables and Models in Python

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates how to define tables and models using both declarative and constructor-style approaches within the ClickHouse SQLAlchemy dialect.  It shows how to define columns with various data types and specify table engine and comments.

```python
from sqlalchemy import create_engine, Column, MetaData, literal

from clickhouse_sqlalchemy import (
    Table, make_session, get_declarative_base, types, engines
)

uri = 'clickhouse://default:@localhost/test'

engine = create_engine(uri)
session = make_session(engine)
metadata = MetaData(bind=engine)

Base = get_declarative_base(metadata=metadata)

class Rate(Base):
    day = Column(types.Date, primary_key=True)
    value = Column(types.Int32, comment='Rate value')
    other_value = Column(types.DateTime)

    __table_args__ = (
        engines.Memory(),
        {'comment': 'Store rates'}
    )

another_table = Table('another_rate', metadata,
    Column('day', types.Date, primary_key=True),
    Column('value', types.Int32, server_default=literal(1)),
    engines.Memory()
)
```

--------------------------------

### ClickHouse Lambda Functions for Array Operations with SQLAlchemy

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows how to use lambda functions for array manipulation in ClickHouse queries via SQLAlchemy. This example uses `arrayFilter` with a lambda function to filter elements within an array based on a condition.

```python
from clickhouse_sqlalchemy.ext.clauses import Lambda
from sqlalchemy import func

# Assuming session and PageViews are defined
# lambda_query = session.query(
#     func.arrayFilter(
#         Lambda(lambda x: x.like('%important%')),
#         PageViews.tags
#     ).label('filtered_tags')
# ).first()
```

--------------------------------

### Create Materialized View with SQLAlchemy and ClickHouse

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows how to define and create Materialized Views in ClickHouse using SQLAlchemy. This includes setting up source and storage tables with appropriate engines (MergeTree, SummingMergeTree) and defining the SELECT query for data aggregation. Supports both Ordinary and Atomic database engines.

```python
from sqlalchemy import create_engine, Column, MetaData, func
from clickhouse_sqlalchemy import (
    Table, make_session, get_declarative_base, types, engines,
    MaterializedView, select
)
import sqlalchemy as sa

engine = create_engine('clickhouse+native://localhost/default')
session = make_session(engine)
metadata = MetaData(bind=engine)
Base = get_declarative_base(metadata=metadata)

# Source table definition
class RawEvents(Base):
    __tablename__ = 'raw_events'
    timestamp = Column(types.DateTime, primary_key=True)
    user_id = Column(types.UInt64)
    event_type = Column(types.String)
    value = Column(types.Int32)

    __table_args__ = (
        engines.MergeTree(
            partition_by=func.toYYYYMM(timestamp),
            order_by=(timestamp, user_id)
        ),
    )

# Storage table for materialized view (Ordinary engine)
class HourlyStats(Base):
    __tablename__ = 'hourly_stats'
    hour = Column(types.DateTime, primary_key=True)
    event_type = Column(types.String)
    event_count = Column(types.UInt64)
    total_value = Column(types.Int64)

    __table_args__ = (
        engines.SummingMergeTree(
            columns=(event_count, total_value),
            partition_by=func.toYYYYMM(hour),
            order_by=(hour, event_type)
        ),
    )

# Create the source and storage tables
RawEvents.__table__.create()
HourlyStats.__table__.create()

# Define materialized view SELECT query
hourly_mv = MaterializedView(
    HourlyStats,
    select(
        func.toStartOfHour(RawEvents.timestamp).label('hour'),
        RawEvents.event_type.label('event_type'),
        func.count().label('event_count'),
        func.sum(RawEvents.value).label('total_value')
    ).group_by(
        func.toStartOfHour(RawEvents.timestamp),
        RawEvents.event_type
    )
)

# Create the materialized view
hourly_mv.create()

# Storage table for materialized view (Atomic engine example)
class DailyStats(Base):
    __tablename__ = 'daily_stats'
    day = Column(types.Date, primary_key=True)
    unique_users = Column(types.UInt64)
    total_events = Column(types.UInt64)

    __table_args__ = (
        engines.SummingMergeTree(
            partition_by=func.toYYYYMM(day),
            order_by=(day,)
        ),
    )

DailyStats.__table__.create()

# MaterializedView with use_to=True for Atomic database
daily_mv = MaterializedView(
    DailyStats,
    select(
        func.toDate(RawEvents.timestamp).label('day'),
        func.uniqExact(RawEvents.user_id).label('unique_users'),
        func.count().label('total_events')
    ).group_by(func.toDate(RawEvents.timestamp)),
    use_to=True,  # Required for Atomic database engine
    name='daily_stats_mv'  # Optional custom view name
)

daily_mv.create()

# Materialized view with AggregateFunction for complex aggregations
class AdvancedStats(Base):
    __tablename__ = 'advanced_stats'
    day = Column(types.Date, primary_key=True)
    event_type = Column(types.String)
    avg_value = Column(types.SimpleAggregateFunction(sa.func.sum, types.Int64))

    __table_args__ = (
        engines.AggregatingMergeTree(
            partition_by=func.toYYYYMM(day),
            order_by=(day, event_type)
        ),
    )

# Insert data into source table - materialized views update automatically
events = [
    {'timestamp': '2024-01-15 10:30:00', 'user_id': 1, 'event_type': 'click', 'value': 10},
    {'timestamp': '2024-01-15 10:45:00', 'user_id': 2, 'event_type': 'view', 'value': 5},
    {'timestamp': '2024-01-15 11:00:00', 'user_id': 1, 'event_type': 'click', 'value': 15},
]
session.execute(RawEvents.__table__.insert(), events)
session.commit()

# Query the materialized view
stats = session.query(HourlyStats).all()
for stat in stats:
    print(f"{stat.hour}: {stat.event_type} - {stat.event_count} events, total: {stat.total_value}")
```

--------------------------------

### ClickHouse WITH TOTALS Clause with SQLAlchemy

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Illustrates the use of the WITH TOTALS clause in conjunction with GROUP BY using SQLAlchemy. This adds an extra row to the result set that contains the aggregated totals across all groups, simplifying the process of getting both group-level and overall summaries.

```python
from sqlalchemy import func

# Assuming session and PageViews are defined
# totals_query = session.query(
#     PageViews.page,
#     func.count().label('views'),
#     func.sum(PageViews.duration).label('total_duration')
# ).group_by(PageViews.page) \
#     .with_totals() \
#     .all()
```

--------------------------------

### Alembic Migration: Create Materialized View (Python)

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Shows how to create a materialized view in ClickHouse using Alembic migrations. This includes creating a storage table and then defining the materialized view with a SELECT query and optional cluster configuration.

```python
from alembic import op

revision = '003'
down_revision = '002'

def upgrade():
    # Create storage table
    op.execute("""
        CREATE TABLE hourly_stats (
            hour DateTime,
            event_count UInt64,
            total_value Int64
        ) ENGINE = SummingMergeTree()
        ORDER BY hour
    """)

    # Create materialized view
    op.create_mat_view(
        'hourly_stats_mv',
        """
        SELECT
            toStartOfHour(event_date) as hour,
            count() as event_count,
            sum(value) as total_value
        FROM events
        GROUP BY hour
        """,
        to='hourly_stats',
        if_not_exists=True,
        on_cluster='my_cluster'  # Optional cluster
    )

def downgrade():
    op.detach_mat_view(
        'hourly_stats_mv',
        if_exists=True,
        permanently=True
    )
    op.drop_table('hourly_stats')
```

--------------------------------

### Define SummingMergeTree Engine (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Illustrates the definition of the SummingMergeTree engine, which requires a `columns` argument specifying which columns to sum. Partitioning and ordering are also configured.

```python
engines.SummingMergeTree(
    columns=(y, ),
    partition_by=date,
    order_by=(date, x)
)
```

--------------------------------

### ClickHouse SQLAlchemy JOIN Operations

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Demonstrates various JOIN operations in ClickHouse using SQLAlchemy, including standard INNER, LEFT OUTER, and FULL OUTER JOINs. It also showcases ClickHouse-specific JOIN types like ANY, ALL, SEMI, ANTI, and GLOBAL JOINs, along with parameters for strictness and distribution for optimized distributed query execution. Requires clickhouse-sqlalchemy and a ClickHouse engine connection.

```python
from sqlalchemy import create_engine, Column, MetaData, func
from clickhouse_sqlalchemy import Table, make_session, get_declarative_base, types, engines, select

engine = create_engine('clickhouse+native://localhost/default')
session = make_session(engine)
metadata = MetaData(bind=engine)
Base = get_declarative_base(metadata=metadata)

class Orders(Base):
    order_id = Column(types.UInt64, primary_key=True)
    user_id = Column(types.UInt64)
    amount = Column(types.Float64)
    __table_args__ = (engines.Memory(),)

class Users(Base):
    user_id = Column(types.UInt64, primary_key=True)
    name = Column(types.String)
    country = Column(types.String)
    __table_args__ = (engines.Memory(),)

# Standard INNER JOIN
inner_join = session.query(
    Orders.order_id,
    Users.name,
    Orders.amount
).join(
    Users,
    Orders.user_id == Users.user_id
).all()

# LEFT OUTER JOIN
left_join = session.query(
    Users.name,
    func.coalesce(func.sum(Orders.amount), 0).label('total')
).outerjoin(
    Orders,
    Users.user_id == Orders.user_id
).group_by(Users.name).all()

# FULL OUTER JOIN
full_join = session.query(
    Users.user_id,
    Orders.order_id
).join(
    Orders,
    Users.user_id == Orders.user_id,
    isouter=True,
    full=True
).all()

# ClickHouse-specific JOIN with strictness (ANY/ALL/ASOF)
# ANY JOIN - return first matching row (faster)
any_join = session.query(
    Orders.order_id,
    Users.name
).join(
    Users,
    Orders.user_id == Users.user_id,
    type='inner',
    strictness='any'  # Return any single match
).all()

# ALL JOIN - return all matching rows (default SQL behavior)
all_join = session.query(
    Orders.order_id,
    Users.name
).join(
    Users,
    Orders.user_id == Users.user_id,
    type='inner',
    strictness='all'
).all()

# GLOBAL JOIN for distributed queries - broadcasts right table to all nodes
global_join = session.query(
    Orders.order_id,
    Users.name
).join(
    Users,
    Orders.user_id == Users.user_id,
    type='inner',
    strictness='all',
    distribution='global'  # GLOBAL ALL INNER JOIN
).all()

# SEMI JOIN - returns left rows that have a match (no right columns)
semi_join = session.query(Orders) \
    .join(
        Users,
        Orders.user_id == Users.user_id,
        strictness='semi'  # LEFT SEMI JOIN
    ).all()

# ANTI JOIN - returns left rows that have NO match
anti_join = session.query(Orders) \
    .join(
        Users,
        Orders.user_id == Users.user_id,
        strictness='anti'  # LEFT ANTI JOIN
    ).all()

# Using select() with joins
stmt = select(Orders.order_id, Users.name, Orders.amount) \
    .join(
        Users,
        Orders.user_id == Users.user_id,
        type='left',
        strictness='any',
        distribution='global'
    )
results = session.execute(stmt).fetchall()
```

--------------------------------

### Special Use Case Engines (Log, Buffer, Memory, Null)

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Showcases simple ClickHouse table engines for specific use cases: Log (append-only), Buffer (temporary storage), Memory (RAM-only), and Null (discards data). Each engine is defined with its specific parameters.

```python
logs_table = Table('logs', metadata,
    Column('message', types.String),
    engines.Log()  # Append-only, no indexes
)

buffer_table = Table('buffer', metadata,
    Column('data', types.String),
    engines.Buffer('default', 'destination_table', 16, 10, 100, 10000, 1000000, 10000000, 100000000)
)

memory_table = Table('temp', metadata,
    Column('value', types.Int32),
    engines.Memory()  # RAM-only, fast but volatile
)

null_table = Table('blackhole', metadata,
    Column('data', types.String),
    engines.Null()  # Discards all data
)
```

--------------------------------

### ClickHouse SQLAlchemy Connection String Syntax

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/connection.md

Defines the general syntax for ClickHouse SQLAlchemy connection strings, specifying the protocol, driver, user, password, host, port, and database. It also outlines default values and how additional parameters are passed to the driver.

```default
clickhouse<+driver>://<user>:<password>@<host>:<port>/<database>[?key=value..]
```

--------------------------------

### Batching Results with yield_per in SQLAlchemy

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Describes how to fetch large result sets in chunks using the yield_per method in SQLAlchemy, which is supported by the native ClickHouse driver. It highlights potential side effects and provides a solution using a separate session for iterations.

```python
session.query(...).yield_per(N)
```

```python
def gen(session):
    yield from session.query(...).yield_per(N)

rv = gen(session)

# There will be an error
session.query(...).all()
```

```python
class another_session():
    def __init__(self, engine):
        self.engine = engine
        self.session = None

    def __enter__(self):
        self.session = make_session(self.engine)
        return self.session

    def __exit__(self, *exc_info):
        self.session.close()

def gen(session):
    with another_session(session.bind) as new_session:
        yield from new_session.query(...).yield_per(N)

rv = gen(session)

# There will be no error
session.query(...).all()
```

--------------------------------

### Limit By Clause in SQLAlchemy

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates how to apply a LIMIT BY clause to a query using SQLAlchemy. This is useful for limiting rows within partitions defined by specific columns. It shows usage with both session.query and select.

```python
session.query(table.c.x).order_by(table.c.x) \
    .limit_by([table.c.x], offset=1, limit=2)
```

```python
select([table.c.x]).order_by(table.c.x) \
    .limit_by([table.c.x], offset=1, limit=2)
```

--------------------------------

### Query Method Chaining (COUNT, ORDER BY, LIMIT)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates common query method chaining patterns supported by the ClickHouse SQLAlchemy dialect. This includes aggregation functions like `count`, ordering with `order_by`, and limiting results with `limit` and `offset`. It also shows how to retrieve a single scalar value or the first result.

```python
session.query(func.count(Rate.day)) \
    .filter(Rate.day > today - timedelta(20)) \
    .scalar()

session.query(Rate.value) \
    .order_by(Rate.day.desc()) \
    .first()

session.query(Rate.value) \
    .order_by(Rate.day) \
    .limit(10) \
    .all()

session.query(func.sum(Rate.value)) \
    .scalar()
```

--------------------------------

### Define CollapsingMergeTree Engine Imperatively (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Illustrates defining a ClickHouse table with the CollapsingMergeTree engine using SQLAlchemy's imperative syntax. This approach involves creating a `Table` object and directly including the engine definition within its arguments. It shows how to use `sqlalchemy.text` for complex expressions.

```python
from sqlalchemy import create_engine, MetaData, Column, text
from clickhouse_sqlalchemy import (
    get_declarative_base, types, engines
)

engine = create_engine('clickhouse+native://localhost/default')
metadata = MetaData(bind=engine)

statistics = Table(
    'statistics', metadata,
    Column('date', types.Date, primary_key=True),
    Column('sign', types.Int8),
    Column('grouping', types.Int32),
    Column('metric1', types.Int32),

    engines.CollapsingMergeTree(
        'sign',
        partition_by=text('toYYYYMM(date)'),
        order_by=('date', 'grouping')
    )
)
```

--------------------------------

### Define VersionedCollapsingMergeTree Engine (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Shows how to define the VersionedCollapsingMergeTree engine, which requires both a `sign` and a `version` column. It also includes parameters for partitioning and ordering.

```python
engines.VersionedCollapsingMergeTree(
    sign, version,
    partition_by=date,
    order_by=(date, x),
)
```

--------------------------------

### Define ReplacingMergeTree Engine (Python)

Source: https://github.com/xzkostyan/clickhouse-sqlalchemy/blob/master/docs/features.md

Demonstrates the definition of the ReplacingMergeTree engine, which uses a `version` column to determine which row to keep. Partitioning and ordering are also specified.

```python
engines.ReplacingMergeTree(
    version='version',
    partition_by='date',
    order_by=('date', 'x')
)
```

--------------------------------

### ClickHouse Table Reflection with SQLAlchemy (Python)

Source: https://context7.com/xzkostyan/clickhouse-sqlalchemy/llms.txt

Illustrates how to reflect existing ClickHouse table schemas into SQLAlchemy models. This allows you to inspect table structure, columns, data types, and query data using familiar SQLAlchemy patterns.

```python
from sqlalchemy import create_engine, MetaData, inspect
from clickhouse_sqlalchemy import Table, make_session

# Standard connection
engine = create_engine('clickhouse+native://localhost/default')
metadata = MetaData(bind=engine)
session = make_session(engine)

# Reflect single table
events = Table('events', metadata, autoload=True)

# Access reflected columns
for column in events.columns:
    print(f"{column.name}: {column.type}")
# Output:
# event_date: Date
# user_id: UInt64
# value: Int32

# Query reflected table
results = session.query(events.c.user_id, events.c.value) \
    .filter(events.c.value > 100) \
    .all()

# Reflect with engine details (slower but complete)
events_full = Table('events', metadata, autoload=True)
print(f"Engine: {events_full.engine}")  # MergeTree details

# Disable engine reflection for faster metadata loading
fast_engine = create_engine(
    'clickhouse+native://localhost/default?engine_reflection=false'
)
fast_metadata = MetaData(bind=fast_engine)
events_fast = Table('events', fast_metadata, autoload=True)

# Use inspector for database introspection
inspector = inspect(engine)

# List all tables
tables = inspector.get_table_names()
print(f"Tables: {tables}")

# List tables in specific schema/database
tables_in_db = inspector.get_table_names(schema='analytics')

# Get column details
columns = inspector.get_columns('events')
for col in columns:
    print(f"{col['name']}: {col['type']} (nullable: {col['nullable']})")

# Get primary key
pk = inspector.get_pk_constraint('events')
print(f"Primary key: {pk}")

# Reflect all tables in database
metadata.reflect()
for table_name in metadata.tables:
    table = metadata.tables[table_name]
    print(f"Table {table_name}: {len(table.columns)} columns")

# Reflect specific tables only
metadata2 = MetaData(bind=engine)
metadata2.reflect(only=['events', 'users', 'orders'])
```
Tokens:
Raw