asynch: asyncio Driver for ClickHouse

For ClickHouse 24.8+, JSON columns are supported. When reading, they are returned as nested Python dictionaries. When inserting, you can provide either a Python dictionary or a JSON-formatted string.

async def use_json(conn: Connection):
    async with conn.cursor() as cursor:
        await cursor.execute(
            "CREATE TABLE test.events (id UInt32, doc JSON) ENGINE = MergeTree ORDER BY id"
        )
        await cursor.execute(
            "INSERT INTO test.events (id, doc) VALUES",
            [
                (1, {"user": {"name": "ada"}, "tags": ["a", "b"]}),
                (2, '{"user": {"name": "bob"}}'),   # JSON text works too
            ],
        )

        await cursor.execute("SELECT doc FROM test.events ORDER BY id")
        assert await cursor.fetchone() == ({"user": {"name": "ada"}, "tags": ["a", "b"]},)

Install the asynch driver using pip. Binary wheels are available for Linux, Windows, and macOS (Intel/ARM) on Python 3.11–3.14. If a wheel is not available, a C toolchain is required for compilation.

To enable transport compression, install with the [compression] extra, which requires clickhouse-cityhash.

> pip install asynch

# With compression support
> pip install asynch[compression]

To avoid loading massive result sets into memory at once, enable streaming on the cursor. This allows you to iterate over rows block-by-block from the server using async for.

async def stream_rows(conn: Connection):
    async with conn.cursor() as cursor:
        # Enable streaming with a specific max_row_buffer
        cursor.set_stream_results(stream_results=True, max_row_buffer=65536)
        await cursor.execute("SELECT number FROM system.numbers LIMIT 1000000")
        async for row in cursor:
            process(row)

The Pool implements several mechanisms to ensure connection health and resource efficiency:

Liveness Checks

To avoid the latency of extra PING round-trips in production, the pool uses a liveness_grace period. If a connection was verified within this grace period, the pool skips the is_live() check during checkout. Otherwise, it performs a liveness check before handing the connection to the user.

Idle Connection Reaping

If idle_timeout is configured, the pool periodically reaps connections that have been idle for longer than the timeout. The reaping process ensures that the number of free connections does not drop below minsize.

Connection Discarding

Connections are automatically discarded (closed) if:

1. They fail a liveness check during checkout.
2. They are found to be dead during the release process.
3. They exceed the idle_timeout during a reaping cycle.

The Connection.last_query attribute provides metadata about the most recent query executed on that connection, such as elapsed time and progress.

async def show_stats(conn: Connection):
    async with conn.cursor() as cursor:
        await cursor.execute("SELECT number FROM system.numbers LIMIT 100000")
        await cursor.fetchall()

    stats = conn.last_query
    print(stats.elapsed, stats.progress.rows, stats.progress.bytes)

Acquire a Cursor from a Connection instance to execute SQL statements. By default, conn.cursor() returns a standard Cursor which returns rows as tuples.

Common methods:

• execute(query, parameters=None): Executes the SQL statement.
• fetchone(): Fetches the next single row.
• fetchall(): Fetches all remaining rows as a list of tuples.

# Fetching one row
async with conn.cursor() as cursor:
    await cursor.execute("SELECT 1")
    ret = await cursor.fetchone()
    assert ret == (1,)

# Fetching all rows
async with conn.cursor() as cursor:
    await cursor.execute("SELECT 1")
    ret = await cursor.fetchall()
    assert ret == [(1,)]

You can stop a long-running query from another task using conn.cancel() or cursor.cancel(). This leaves the connection usable for subsequent operations.

import asyncio

async def cancel_slow_query(conn: Connection):
    async with conn.cursor() as cursor:
        # Start a long query in a separate task
        task = asyncio.create_task(cursor.execute("SELECT count() FROM numbers(20000000000)"))
        await asyncio.sleep(1)
        
        # Cancel the query via the connection or cursor
        await conn.cancel() 
        
        # The task will return with whatever the server had sent before cancellation
        await task

You can establish a connection to a ClickHouse server using the Connection class in two ways:

1. DSN String: Use a connection string in the format clickhouse://[user:password]@host:port/database. If a DSN is provided, it takes priority over individual parameters.
2. DSN Parameters: Provide user, password, host, port, and database as separate arguments.

Always use async with to ensure the connection is properly closed.

from asynch import Connection

# Using a DSN string
async def connect_with_dsn():
    async with Connection(dsn="clickhouse://ch_user:P%4055w0rD@127.0.0.1:9000/chdb") as conn:
        pass

# Using separate parameters
async def connect_with_params():
    async with Connection(
        user="ch_user",
        password="P@55w0rD",
        host="127.0.0.1",
        port=9000,
        database="chdb",
    ) as conn:
        pass