aiochlite

Lightweight asynchronous ClickHouse client for Python built on aiohttp.

Table of Contents

• Features
• Why aiochlite?
• Installation
• Quick Start
  • Basic Connection
  • Execute Query
  • Insert Data
  • Fetch Results
  • Export Formats
  • Query Parameters
  • Query Settings
  • External Tables
  • JSON Type
  • Error Handling
  • Custom Session
  • Enable Compression
• Type Conversion
• Benchmarks
• License

Features

• Lightweight - minimal dependencies, only aiohttp required
• Streaming support - efficient processing of large datasets with .stream()
• Export formats - raw Parquet / CSV / TSV / JSON / Arrow / ORC payloads via .fetch_format() and .stream_format()
• External tables - advanced temporary data support
• Type conversion - automatic conversion between Python and ClickHouse types
• Type-safe - full type hints coverage
• Flexible - custom sessions, compression, query settings

Why aiochlite?

• Real asyncio I/O: built on aiohttp without wrapping blocking code in a thread pool.
• Fast decoding: uses RowBinaryWithNamesAndTypes and lets you choose between Row wrappers (fetch()) and raw tuples (fetch_rows()).
• Small surface area: minimal dependencies and a focused API for ClickHouse HTTP.

Installation

pip install aiochlite

Quick Start

Basic Connection

from aiochlite import AsyncChClient

# Using context manager (recommended)
async with AsyncChClient(
    url="http://localhost:8123",
    user="default",
    password="",
    database="default"
) as client:
    result = await client.fetch("SELECT 1")

# Or manual connection management
client = AsyncChClient("http://localhost:8123")
try:
    assert await client.ping()
    result = await client.fetch("SELECT 1")
finally:
    await client.close()

Execute Query

await client.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id UInt32,
        name String,
        email String
    ) ENGINE = MergeTree() ORDER BY id
""")

Insert Data

# Insert dictionaries
data = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"},
]
await client.insert("users", data)

# Insert tuples
data = [
    (3, "Charlie", "charlie@example.com"),
    (4, "Diana", "diana@example.com"),
]
await client.insert("users", data, column_names=["id", "name", "email"])

# Insert with settings
await client.insert(
    "users",
    [{"id": 5, "name": "Eve", "email": "eve@example.com"}],
    settings={"max_insert_block_size": 100000}
)

Fetch Results

# Fetch all rows
rows = await client.fetch("SELECT * FROM users")
for row in rows:
    print(f"ID: {row.id}, Name: {row.name}, Email: {row.email}")

# Fetch one row
row = await client.fetchone("SELECT * FROM users WHERE id = 1")
if row:
    print(row.name)  # Attribute access
    print(row["name"])  # Dictionary-style access
    print(row.first())  # Get first column value

# Fetch single value
count = await client.fetchval("SELECT count() FROM users")
print(f"Total users: {count}")

# Iterate over results (for large datasets)
async for row in client.stream("SELECT * FROM users"):
    print(row.name)

Export Formats

Get the raw server payload in any ClickHouse output format — Parquet, CSV, TSV, JSON, Arrow, ORC and more. Both methods return raw bytes exactly as produced by the server (decode text formats yourself).

# Whole result at once
parquet = await client.fetch_format("SELECT * FROM users", "Parquet")
csv = (await client.fetch_format("SELECT * FROM users", "CSVWithNames")).decode()

# Chunked streaming for large result sets
with open("users.parquet", "wb") as f:
    async for chunk in client.stream_format("SELECT * FROM users", "Parquet"):
        f.write(chunk)

# Query parameters, settings and external tables work as usual
ndjson = await client.fetch_format(
    "SELECT * FROM users WHERE id > {id:UInt32}",
    "JSONEachRow",
    params={"id": 10},
)

Supported formats (ExportFormat):

Any other output format the server accepts can still be passed at runtime (type checkers will flag it).

[!WARNING] fetch_parquet() and stream_parquet() are deprecated and will be removed in a future release. Use fetch_format(query, "Parquet") and stream_format(query, "Parquet") instead.

Query Parameters

# Basic types
result = await client.fetch(
    "SELECT * FROM users WHERE id = {id:UInt32}",
    params={"id": 1}
)

# Lists and tuples (arrays)
result = await client.fetch(
    "SELECT * FROM users WHERE id IN {ids:Array(UInt32)}",
    params={"ids": [1, 2, 3]}  # or tuple: (1, 2, 3)
)

# Datetime and date
from datetime import datetime, date

result = await client.fetch(
    "SELECT * FROM events WHERE created_at > {dt:DateTime} AND date = {d:Date}",
    params={
        "dt": datetime(2025, 12, 14, 15, 30, 45),
        "d": date(2025, 12, 14)
    }
)

# UUID
from uuid import UUID

result = await client.fetch(
    "SELECT * FROM users WHERE uuid = {uid:UUID}",
    params={"uid": UUID("550e8400-e29b-41d4-a716-446655440000")}
)

# Decimal
from decimal import Decimal

result = await client.fetch(
    "SELECT * FROM products WHERE price > {price:Decimal(10, 2)}",
    params={"price": Decimal("99.99")}
)

# Nested arrays and maps
result = await client.fetch(
    "SELECT {matrix:Array(Array(Int32))} AS matrix, {data:Map(String, Int32)} AS data",
    params={
        "matrix": [[1, 2], [3, 4]],
        "data": {"a": 1, "b": 2}
    }
)

Supported parameter types:

• Basic: int, float, str, bool, None
• Collections: list, tuple, dict
• Date/Time: datetime, date, timedelta
• Special: UUID, Decimal, bytes

See Type Conversion for full type mapping details.

Query Settings

rows = await client.fetch(
    "SELECT * FROM users",
    settings={
        "max_execution_time": 60,
        "max_block_size": 10000
    }
)

External Tables

from aiochlite import ExternalTable

external_data = {
    "temp_data": ExternalTable(
        structure=[("id", "UInt32"), ("value", "String")],
        data=[
            {"id": 1, "value": "foo"},
            {"id": 2, "value": "bar"},
        ]
    )
}

result = await client.fetch(
    """
    SELECT t1.id, t1.name, t2.value
    FROM users t1
    JOIN temp_data t2 ON t1.id = t2.id
    """,
    external_tables=external_data
)

JSON Type

[!NOTE] For ClickHouse versions where JSON is still considered experimental, set allow_experimental_json_type=1 via client settings.

await client.execute("DROP TABLE IF EXISTS json_demo")
await client.execute("CREATE TABLE json_demo (id UInt32, doc JSON) ENGINE = Memory")

await client.insert(
    "json_demo",
    [{"id": 1, "doc": {"a": 1, "b": [True, None, {"c": "x"}]}}],
)

row = await client.fetchone("SELECT id, doc FROM json_demo WHERE id = 1")
print(row["doc"])  # Output: {"a": 1, "b": [True, None, {"c": "x"}]}

Error Handling

from aiochlite import ChClientError

try:
    await client.execute("SELECT * FROM non_existent_table")
except ChClientError as e:
    print(f"Query failed: {e}")

Custom Session

from aiohttp import ClientSession, ClientTimeout

timeout = ClientTimeout(total=30)
async with ClientSession(timeout=timeout) as session:
    async with AsyncChClient(url="http://localhost:8123", session=session) as client:
        result = await client.fetch("SELECT 1")

Enable Compression

async with AsyncChClient(url="http://localhost:8123", enable_compression=True) as client:
    result = await client.fetch("SELECT * FROM users")

Type Conversion

aiochlite uses ClickHouse’s RowBinaryWithNamesAndTypes for result decoding:

• fetch, fetchone, fetchval, stream automatically append FORMAT RowBinaryWithNamesAndTypes and decode rows into Python values.
• Queries passed to these methods must not contain a FORMAT ... clause.
• Use execute() for statements that don’t return rows.

Automatic type conversion from ClickHouse:

Python to ClickHouse conversion:

When sending data to ClickHouse (query parameters and inserts), Python types are automatically converted:

• datetime → YYYY-MM-DD HH:MM:SS
• date → YYYY-MM-DD
• timedelta → HH:MM:SS[.ffffff] (signed; suitable for Time / Time64)
• UUID / Decimal → string representation
• list → array literal (e.g. [1,2,3])
• tuple → tuple literal (e.g. (1,2,3))
• dict → map literal (e.g. {'k':'v'})
• bytes → UTF-8 decoded string
• None → NULL
• bool → 1/0 for query parameters, true/false inside container literals

Benchmarks

Benchmark scripts live in benchmarks/.

[!NOTE] Benchmarks always depend on machine and environment (CPU, RAM, kernel, ClickHouse version/config, network, etc). These results were captured on a local machine with 8 CPU cores (16 threads) and 32 GB RAM, running ClickHouse 26.3 LTS. Each client uses the configuration recommended by its documentation, including the aiohttp-speedups extra for aiochclient.

Latest fetch-and-decode results for 100,000 rows (5 rounds, measured 2026-08-14):

Versions: aiochlite 1.3.0, clickhouse-connect 1.7.1, aiochclient 2.7.0, Python 3.14.5, and ClickHouse 26.3.17.110.

clickhouse-connect includes compiled C extensions. In contrast, aiochlite is pure Python and has a single dependency: aiohttp.

License

MIT License

Copyright (c) 2025 darkstussy