Asynchronous ClickHouse Driver (asynch)

Warnings

• breaking The top-level `asynch.connect()` function has been removed. Direct instantiation and usage of `Connection` within an `async with` statement is now required.
  Fix: Replace `await asynch.connect(...)` with `async with Connection(...) as conn:`.
• breaking The `Connection.connected` property, used to check connection status, has been renamed.
  Fix: Use `Connection.opened` instead of `Connection.connected`.
• breaking The `asynch.pool.create_pool()` function has been removed. Connection pooling should now be managed directly using the `Pool` class.
  Fix: Replace `await create_pool(...)` with `async with Pool(...) as pool:`.
• gotcha Always use `async with Connection(...)` and `async with conn.cursor(...)` to ensure connections and cursors are properly managed and closed. Failing to do so can lead to resource leaks and unclosed connections, especially under heavy load.
  Fix: Wrap `Connection` and cursor operations in `async with` statements to leverage asynchronous context management.
• gotcha Blocking the asyncio event loop with synchronous I/O or long-running CPU-bound tasks within an async function can severely degrade performance and responsiveness of your application. Ensure all I/O operations are truly asynchronous or offloaded to a thread/process pool.
  Fix: Identify and refactor blocking calls. Use `await` with async-compatible libraries (like `asynch`) and for CPU-bound tasks, consider `asyncio.to_thread()` or `loop.run_in_executor()`.

Install

• pip install asynch Basic Installation
• pip install asynch[compression] With `clickhouse-cityhash` for compression

Imports

• Connection

  from asynch import Connection

• Pool

  from asynch import Pool

• Cursor

  from asynch.cursors import Cursor

  Cursor classes are found in the `asynch.cursors` submodule.
• DictCursor

  from asynch.cursors import DictCursor

  Cursor classes are found in the `asynch.cursors` submodule.
• connect

  from asynch import Connection # Use Connection class directly

  The `connect` function was removed in v0.3.0; use `async with Connection(...)` instead.
• create_pool

  from asynch import Pool # Use Pool class directly

  The `create_pool` function was removed in v0.3.0; use `async with Pool(...)` instead.

Quickstart

This quickstart demonstrates how to establish a connection to a ClickHouse server, create a table, insert data, and fetch results using `asynch`. It showcases the use of `async with` for managing connections and cursors, including `DictCursor` for dictionary-like row access. Environment variables are used for connection details for security.

import asyncio
from asynch import Connection
from asynch.cursors import DictCursor
import os

async def main():
    # Connect using DSN parameters
    async with Connection(
        user=os.environ.get('CH_USER', 'default'),
        password=os.environ.get('CH_PASSWORD', ''),
        host=os.environ.get('CH_HOST', '127.0.0.1'),
        port=int(os.environ.get('CH_PORT', 9000)),
        database=os.environ.get('CH_DATABASE', 'default'),
    ) as conn:
        print(f"Connected to ClickHouse: {conn.opened}")

        # Create a table
        async with conn.cursor() as cursor:
            await cursor.execute(
                "CREATE TABLE IF NOT EXISTS test_table (id UInt64, value String) ENGINE = MergeTree ORDER BY id"
            )
            print("Table 'test_table' created or already exists.")

            # Insert data
            await cursor.execute(
                "INSERT INTO test_table (id, value) VALUES",
                [[1, 'hello'], [2, 'world']]
            )
            print("Data inserted.")

        # Select data using DictCursor
        async with conn.cursor(cursor=DictCursor) as cursor:
            await cursor.execute("SELECT id, value FROM test_table ORDER BY id")
            result = await cursor.fetchall()
            print(f"Fetched data: {result}")
            
        # Clean up (optional)
        async with conn.cursor() as cursor:
            await cursor.execute("DROP TABLE IF EXISTS test_table")
            print("Table 'test_table' dropped.")

if __name__ == "__main__":
    # Example environment variables (replace with your ClickHouse instance details)
    # os.environ['CH_USER'] = 'my_user'
    # os.environ['CH_PASSWORD'] = 'my_password'
    # os.environ['CH_HOST'] = 'localhost'
    # os.environ['CH_PORT'] = '9000'
    # os.environ['CH_DATABASE'] = 'my_db'
    
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"An error occurred: {e}")