aioch

You can install aioch using pip for a standard installation, or by cloning the repository and running the setup script for development/source installation.

# Via pip
pip install aioch

# From source
git clone https://github.com/mymarilyn/aioch
cd aioch
python setup.py install

Use Client.execute_with_progress(query) to run a query while monitoring its progress. This method returns an asynchronous iterator that yields (num_rows, total_rows).

To retrieve the actual result of the query after the iterator is exhausted, call await progress.get_result() within an else block of the async for loop to ensure the query completed successfully.

from aioch import Client
import asyncio

async def exec_progress():
    client = Client('localhost')
    # Returns an async iterator yielding (num_rows, total_rows)
    progress = await client.execute_with_progress('SELECT * FROM my_table')

    async for num_rows, total_rows in progress:
        print(f"Progress: {num_rows}/{total_rows}")
    else:
        # Retrieve the result after completion
        rv = await progress.get_result()
        print(rv)

asyncio.run(exec_progress())

For standard asynchronous query execution without progress monitoring, use Client.execute(query). This method wraps the underlying clickhouse-driver functionality.

from aioch import Client
import asyncio

async def exec_no_progress():
    client = Client('localhost')
    rv = await client.execute('SELECT * FROM my_table')
    print(rv)

asyncio.run(exec_no_progress())

The Client constructor accepts the following parameters:

• executor: An instance of a custom Executor. If not supplied, a default executor is used.
• loop: An asyncio compatible event loop.

Additionally, any parameters accepted by the underlying clickhouse-driver's Client can be passed to the aioch.Client constructor.

Use execute_iter to execute a query and receive an IterQueryResult. This is useful for streaming large result sets row-by-row without loading everything into memory at once.

Parameters:

• query: The SQL query string.
• params: Dictionary of parameters.
• with_column_types: Boolean, if True, includes column types.
• external_tables: Data for external tables.
• query_id: Unique identifier for the query.
• settings: ClickHouse query settings.
• types_check: Boolean for type checking.

async def stream_query(client):
    async for row in await client.execute_iter('SELECT * FROM large_table'):
        print(row)

Use disconnect() to gracefully close the connection to the ClickHouse server.

await client.disconnect()

Use execute_with_progress to run a query and receive a ProgressQueryResult. This allows you to monitor the progress of the query execution.

Parameters:

• query: The SQL query string.
• params: Dictionary of parameters for substitution.
• with_column_types: Boolean, if True, includes column types in the result.
• external_tables: Data for external tables.
• query_id: Unique identifier for the query.
• settings: ClickHouse query settings.
• types_check: Boolean for type checking.
• columnar: Boolean to return results in columnar format.

async def run_query(client):
    result = await client.execute_with_progress(
        'SELECT * FROM my_table WHERE id = %(id)s',
        params={'id': 123},
        with_column_types=True
    )
    # result is a ProgressQueryResult
    async for row in result:
        print(row)

The Client class provides an asynchronous interface for ClickHouse queries. You can initialize it in two ways:

1. Using from_url: The recommended way to connect via a connection string.
2. Passing an existing _client: If you already have a clickhouse_driver.Client instance, you can wrap it.

You can optionally provide an event_loop or an executor to manage how blocking calls are run in the background.

from aioch import Client

# Recommended: Connect via URL
client = Client.from_url('clickhouse://user:password@host:port/database')

# Optional: Provide custom loop or executor
client = Client.from_url('clickhouse://...', loop=my_loop, executor=my_executor)

You can cancel a query currently being processed by calling cancel().

Note: After sending the cancel signal, the client must still read until the END_OF_STREAM packet is received to ensure the connection is left in a clean state. The method returns the result of the remaining stream (optionally with column types).

# To cancel a query
await client.cancel(with_column_types=True)

Use execute() to run a query and wait for the full result set to be returned. This is a wrapper around the underlying blocking client's execute method, run in an executor.

results = await client.execute('SELECT 1')