API do driver ClickHouse Connect

Use clickhouse_connect.get_client para criar um Client síncrono ou instale o extra async e use await em clickhouse_connect.get_async_client para criar um AsyncClient nativo.

A função de fábrica assíncrona também aceita connector_limit=100, connector_limit_per_host=20 e keepalive_timeout=30.0 para configurar o pool de conexões do aiohttp. Ela não aceita pool_mgr. O backend chDB síncrono aceita path e chdb_options; consulte backend embutido do chDB.

Por fim, o argumento settings de get_client é usado para enviar configurações adicionais do ClickHouse ao servidor em cada solicitação do cliente. Observe que, na maioria dos casos, usuários com acesso readonly=1 não podem alterar configurações enviadas com uma consulta; por isso, o ClickHouse Connect descarta essas configurações na solicitação final e registra um aviso. As configurações a seguir se aplicam apenas a consultas/sessões HTTP usadas pelo ClickHouse Connect e não são documentadas como configurações gerais do ClickHouse.

Para outras configurações do ClickHouse que podem ser enviadas com cada consulta, consulte a documentação do ClickHouse.

• Sem nenhum parâmetro, um cliente ClickHouse Connect se conectará à porta HTTP padrão em localhost, com o usuário default e sem senha:

import clickhouse_connect

client = clickhouse_connect.get_client()
print(client.server_version)

• Conectar-se a um servidor ClickHouse externo seguro (HTTPS)

import clickhouse_connect

client = clickhouse_connect.get_client(
    host="play.clickhouse.com",
    secure=True,
    port=443,
    username="play",
    password="clickhouse",
)
print(client.command("SELECT timezone()"))

• Conectar-se com um ID de sessão e outros parâmetros de conexão personalizados e configurações do ClickHouse.

import clickhouse_connect

client = clickhouse_connect.get_client(
    host="play.clickhouse.com",
    username="play",
    password="clickhouse",
    port=443,
    secure=True,
    session_id="example_session_1",
    connect_timeout=15,
    database="github",
    settings={"distributed_ddl_task_timeout": 300},
)
print(client.database)
# Output: github

Instale clickhouse-connect[chdb] para usar o backend experimental do chDB no próprio processo. Ele disponibiliza os métodos síncronos do cliente para consulta, insert, streaming e Arrow:

import clickhouse_connect

with clickhouse_connect.get_client(interface="chdb") as client:
    result = client.query("SELECT sum(number) FROM numbers(10)")
    print(result.first_row)
    # Output: (45,)

O padrão é um banco de dados em memória. Passe path="/data/my_chdb" ou use dsn="chdb:///data/my_chdb" para armazenamento persistente. O backend permite apenas um caminho de engine por processo e não oferece suporte a get_async_client nem a dados externos.

Criar um cliente ClickHouse Connect é uma operação custosa, que envolve estabelecer uma conexão, recuperar metadados do servidor e inicializar configurações. Siga estas boas práticas para ter o melhor desempenho:

• Reutilize clientes: Crie os clientes uma única vez na inicialização da aplicação e reutilize-os durante todo o seu ciclo de vida
• Evite criação frequente: Não crie um novo cliente para cada consulta ou solicitação
• Faça a limpeza corretamente: Sempre feche os clientes ao encerrar a aplicação para liberar os recursos do pool de conexões
• Compartilhe quando possível: Um único cliente pode processar muitas consultas simultâneas por meio do seu pool de conexões (veja as observações sobre threads abaixo)

Reutilize um único cliente:

import clickhouse_connect

# Create once at startup
client = clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
)

# Reuse for all queries
for i in range(1000):
    result = client.query("SELECT count() FROM users")

# Close on shutdown
client.close()

Evite criar clientes repetidamente:

# BAD: Creates 1000 clients with expensive initialization overhead
for i in range(1000):
    client = clickhouse_connect.get_client(
        host="my-host",
        username="default",
        password="password",
    )
    result = client.query("SELECT count() FROM users")
    client.close()

Para compartilhar um Client entre threads com segurança:

import clickhouse_connect
import threading

# Option 1: Disable sessions (recommended for shared clients)
client = clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
    autogenerate_session_id=False,
)

def worker(thread_id):
    # All threads can now safely use the same client
    result = client.query(f"SELECT {thread_id}")
    print(f"Thread {thread_id}: {result.result_rows[0][0]}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

client.close()

Alternativa para sessões: Se você precisar de sessões (por exemplo, para tabelas temporárias), crie um Client separado para cada thread:

def worker(thread_id):
    # Each thread gets its own client with isolated session
    client = clickhouse_connect.get_client(
        host="my-host",
        username="default",
        password="password",
    )
    client.command("CREATE TEMPORARY TABLE temp (id UInt32) ENGINE = Memory")
    # ... use temp table ...
    client.close()

Sempre feche os clientes ao encerrar. Observe que client.close() descarta o cliente e fecha as conexões HTTP do pool apenas quando o cliente tem seu próprio gerenciador de pool (por exemplo, quando é criado com opções personalizadas de TLS/proxy). Para o pool compartilhado padrão, use client.close_connections() para liberar os sockets de forma proativa; caso contrário, as conexões são liberadas automaticamente por expiração por inatividade e ao encerrar o processo.

client = clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
)
try:
    result = client.query("SELECT 1")
finally:
    client.close()

Ou use um gerenciador de contexto:

with clickhouse_connect.get_client(
    host="my-host",
    username="default",
    password="password",
) as client:
    result = client.query("SELECT 1")

Vários clientes são apropriados para:

• Servidores diferentes: um cliente por servidor ClickHouse ou cluster
• Credenciais diferentes: clientes separados para diferentes usuários ou níveis de acesso
• Bancos de dados diferentes: quando você precisa trabalhar com vários bancos de dados
• Sessões isoladas: quando você precisa de sessões separadas para tabelas temporárias ou configurações específicas da sessão
• Isolamento por thread: quando as threads precisam de sessões independentes (como mostrado acima)

Vários métodos do cliente usam um ou ambos os argumentos comuns parameters e settings. Esses argumentos nomeados são descritos abaixo.

Os métodos query* e command do cliente ClickHouse Connect aceitam o argumento nomeado opcional parameters, usado para vincular expressões Python a uma expressão de valor do ClickHouse. Há dois tipos de vinculação disponíveis.

O ClickHouse oferece vinculação no servidor para valores da consulta. O valor vinculado é enviado separadamente da consulta como um parâmetro HTTP. O ClickHouse Connect usa esse modo quando detecta uma expressão no formato {<name>:<datatype>}. Passe os valores como um dicionário Python.

Os nomes dos parâmetros devem ser nomes ASCII BareWord do ClickHouse. O driver aceita $ no início, no meio ou no final do nome quando o servidor também o aceita, como em {$tenant_id:String}. Uma chave de dicionário que começa e termina com $ e tem um valor de buffer, como bytes, bytearray ou memoryview, é reservada para a convenção de parâmetros binários brutos do ClickHouse Connect. Se essa chave for usada para um parâmetro no servidor que não seja binário, mantenha-a em um único placeholder {name:Type}. Nomes $tag$ repetidos podem ser interpretados pelo ClickHouse como marcadores Heredoc.

Use None do Python para valores anuláveis. Valores None aninhados são compatíveis dentro de parâmetros Array e Tuple e dentro de literais Map quando dict_parameter_format está definido como "map".

• Vinculação no servidor com dicionário Python, valor DateTime e valor String

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {
    "table": "my_table",
    "v1": my_date,
    "v2": "a string with a single quote'",
}
client.query(
    "SELECT * FROM {table:Identifier} "
    "WHERE date >= {v1:DateTime} AND string ILIKE {v2:String}",
    parameters=parameters,
)

Isso equivale a:

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

O ClickHouse Connect também oferece suporte à vinculação de parâmetros no lado do cliente, o que pode proporcionar mais flexibilidade na geração de consultas SQL com template. Para a vinculação no lado do cliente, o argumento parameters deve ser um dicionário ou uma sequência. A vinculação no lado do cliente usa a formatação de strings no estilo "printf" do Python para a substituição de parâmetros.

Observe que, diferentemente da vinculação no servidor, a vinculação no lado do cliente não funciona para identificadores de banco de dados, como nomes de database, table ou coluna, já que a formatação no estilo Python não consegue distinguir entre os diferentes types de strings, e elas precisam ser formatadas de maneira diferente (backticks ou aspas duplas para identificadores de banco de dados, aspas simples para valores de dados).

• Exemplo com Dicionário Python, valor DateTime e escape de string

import datetime

my_date = datetime.datetime(2022, 10, 1, 15, 20, 5)

parameters = {"v1": my_date, "v2": "a string with a single quote'"}
client.query(
    "SELECT * FROM my_table "
    "WHERE date >= %(v1)s AND string ILIKE %(v2)s",
    parameters=parameters,
)

Isso gera a seguinte consulta no servidor:

SELECT *
FROM my_table
WHERE date >= '2022-10-01 15:20:05'
  AND string ILIKE 'a string with a single quote\''

• Exemplo com Sequence do Python (Tuple), Float64 e IPv4Address

import ipaddress

parameters = (35200.44, ipaddress.IPv4Address(0x443d04fe))
client.query(
    "SELECT * FROM some_table WHERE metric >= %s AND ip_address = %s",
    parameters=parameters,
)

Isso gera a seguinte consulta no servidor:

SELECT *
FROM some_table
WHERE metric >= 35200.44
  AND ip_address = '68.61.4.254'

Todos os principais métodos "insert" e "select" do cliente ClickHouse Connect aceitam um argumento nomeado opcional settings para passar configurações de usuário do servidor ClickHouse para a instrução SQL correspondente. O argumento settings deve ser um dicionário. Cada item deve conter o nome de uma configuração do ClickHouse e o respectivo valor. Observe que os valores serão convertidos em strings quando enviados ao servidor como parâmetros de consulta.

Assim como acontece com as configurações no nível do cliente, o ClickHouse Connect descartará todas as configurações que o servidor marcar como readonly=1, com a respectiva mensagem de log. As configurações que se aplicam apenas a consultas feitas pela interface HTTP do ClickHouse são sempre válidas. Essas configurações são descritas na API get_client.

Exemplo de uso das configurações do ClickHouse:

settings = {
    "merge_tree_min_rows_for_concurrent_read": 65535,
    "session_id": "session_1234",
    "use_skip_indexes": False,
}
client.query(
    "SELECT event_type, sum(timeout) "
    "FROM event_errors WHERE event_time > '2022-08-01'",
    settings=settings,
)

Use Client.command para instruções que não retornam um conjunto de dados tabular ou para consultas que retornam um valor primitivo ou uma linha. Dependendo da resposta, ele retorna uma string, um inteiro, uma sequência de strings ou QuerySummary. Uma leitura que produz um conjunto de resultados vazio retorna uma string vazia.

import clickhouse_connect

client = clickhouse_connect.get_client()

# Create a table. A successful DDL returns QuerySummary.
summary = client.command(
    "CREATE TABLE test_command "
    "(col_1 String, col_2 DateTime) "
    "ENGINE MergeTree ORDER BY tuple()"
)
print(summary.query_id())

# Show table definition
result = client.command("SHOW CREATE TABLE test_command")
print(result)
# Output:
# CREATE TABLE default.test_command
# (
#     `col_1` String,
#     `col_2` DateTime
# )
# ENGINE = MergeTree
# ORDER BY tuple()

# Drop table
client.command("DROP TABLE test_command")

import clickhouse_connect

client = clickhouse_connect.get_client()

# Single value result
count = client.command("SELECT count() FROM system.tables")
print(count)

# Server version
version = client.command("SELECT version()")
print(version)

import clickhouse_connect

client = clickhouse_connect.get_client()

# Usando parâmetros no cliente
table_name = "system"
result = client.command(
    "SELECT count() FROM system.tables WHERE database = %(db)s",
    parameters={"db": table_name}
)

# Usando parâmetros no servidor
result = client.command(
    "SELECT count() FROM system.tables WHERE database = {db:String}",
    parameters={"db": "system"}
)

import clickhouse_connect

client = clickhouse_connect.get_client()

# Executa comando com configurações específicas
result = client.command(
    "OPTIMIZE TABLE large_table FINAL",
    settings={"optimize_throw_if_noop": 1}
)

Client.query recupera um conjunto de dados tabular no formato Native do ClickHouse e retorna um QueryResult. O resultado completo é materializado quando uma propriedade do resultado é acessada. Use um método de streaming para resultados que não devem ser mantidos na memória.

import clickhouse_connect

client = clickhouse_connect.get_client()

# Simple SELECT query
result = client.query(
    "SELECT number, toString(number) AS label FROM numbers(3)"
)

# Access results as rows
for row in result.result_rows:
    print(row)
# Output:
# (0, '0')
# (1, '1')
# (2, '2')

# Access column names and types
print(result.column_names)
# Output: ('number', 'label')
print([col_type.name for col_type in result.column_types])
# Output: ['UInt64', 'String']

import clickhouse_connect

client = clickhouse_connect.get_client()

result = client.query("SELECT number, toString(number) AS str FROM system.numbers LIMIT 3")

# Row-oriented access (default)
print(result.result_rows)
# Output: [(0, '0'), (1, '1'), (2, '2')]

# Column-oriented access
print(result.result_columns)
# Output: [[0, 1, 2], ['0', '1', '2']]

# Named results (list of dictionaries)
for row_dict in result.named_results():
    print(row_dict)
# Output:
# {'number': 0, 'str': '0'}
# {'number': 1, 'str': '1'}
# {'number': 2, 'str': '2'}

# First row as dictionary
print(result.first_item)
# Output: {'number': 0, 'str': '0'}

# First row as tuple
print(result.first_row)
# Output: (0, '0')

import clickhouse_connect

client = clickhouse_connect.get_client()

# Usando parâmetros de dicionário (no estilo printf)
query = "SELECT * FROM system.tables WHERE database = %(db)s AND name LIKE %(pattern)s"
parameters = {"db": "system", "pattern": "%query%"}
result = client.query(query, parameters=parameters)

# Usando parâmetros de tupla
query = "SELECT * FROM system.tables WHERE database = %s LIMIT %s"
parameters = ("system", 5)
result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# Binding server-side (mais seguro, melhor desempenho para consultas SELECT)
query = "SELECT * FROM system.tables WHERE database = {db:String} AND name = {tbl:String}"
parameters = {"db": "system", "tbl": "query_log"}

result = client.query(query, parameters=parameters)

import clickhouse_connect

client = clickhouse_connect.get_client()

# Passar as configurações do ClickHouse junto com a consulta
result = client.query(
    "SELECT sum(number) FROM numbers(1000000)",
    settings={
        "max_block_size": 100000,
        "max_execution_time": 30
    }
)

O método base query retorna um objeto QueryResult com as seguintes propriedades públicas:

• result_rows – Matriz de resultados orientada por linhas.
• result_columns – Matriz de resultados orientada por colunas.
• result_set – result_rows ou result_columns, de acordo com a orientação da consulta.
• column_names – Tupla com os nomes das colunas do resultado.
• column_types – Tupla de objetos ClickHouseType.
• row_count – Número de linhas de resultado materializadas.
• query_id – ID da consulta informado ou gerado para a requisição. Uma string vazia significa que nenhum estava disponível.
• summary – Dicionário decodificado do cabeçalho de resposta X-ClickHouse-Summary.
• first_item – Primeira linha como um dicionário, ou None para um resultado vazio.
• first_row – Primeira linha como uma sequência, ou None para um resultado vazio.
• column_block_stream, row_block_stream e rows_stream – Contextos internos de streaming. Use os métodos de streaming correspondentes do cliente.

Consulte Consultas em streaming para ver as APIs StreamContext compatíveis.

O ClickHouse Connect fornece métodos de consulta especializados para os formatos de dados NumPy, Pandas e Arrow. Para informações detalhadas sobre como usar esses métodos, incluindo exemplos, recursos de streaming e tratamento avançado de tipos, consulte Consultas avançadas (consultas com NumPy, Pandas e Arrow).

Para transmitir grandes conjuntos de resultados, o ClickHouse Connect oferece vários métodos de streaming. Consulte Consultas avançadas (consultas em streaming) para mais detalhes e exemplos.

Para o caso de uso comum de inserir vários registros no ClickHouse, existe o método Client.insert. Ele aceita os seguintes parâmetros:

Esse método retorna QuerySummary. Seu dicionário summary contém valores informados pelo servidor. written_rows é uma propriedade de conveniência, enquanto written_bytes() e query_id() retornam os valores correspondentes. Uma falha na inserção gera uma exceção.

Para métodos especializados de inserção que funcionam com Pandas DataFrames, tabelas PyArrow e DataFrames com Arrow como backend, consulte Advanced Inserting (Specialized Insert Methods).

Os exemplos abaixo partem do pressuposto de que já existe uma tabela users com o esquema (id UInt32, name String, age UInt8).

import clickhouse_connect

client = clickhouse_connect.get_client()

# Row-oriented data: each inner list is a row
data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

client.insert("users", data, column_names=["id", "name", "age"])

import clickhouse_connect

client = clickhouse_connect.get_client()

# Column-oriented data: each inner list is a column
data = [
    [13, 79],  # id column
    ["user_1", "user_2"],  # name column
    [25, 30],  # age column
]

client.insert("users", data, column_names=["id", "name", "age"], column_oriented=True)

import clickhouse_connect

client = clickhouse_connect.get_client()

# Useful when you want to avoid a DESCRIBE query to the server
data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    column_type_names=["UInt32", "String", "UInt8"],
)

import clickhouse_connect

client = clickhouse_connect.get_client()

data = [
    [13, "user_1", 25],
    [79, "user_2", 30],
]

# Insert into a table in a specific database
client.insert(
    "users",
    data,
    column_names=["id", "name", "age"],
    database="production",
)

Para inserir dados diretamente de arquivos em tabelas do ClickHouse, consulte Inserção avançada (Inserções a partir de arquivos).

Para casos de uso avançados que exigem acesso direto às interfaces HTTP do ClickHouse, sem transformações de tipo, consulte Uso avançado (API bruta).

O módulo clickhouse_connect.dbapi implementa a interface de conexão e cursor do PEP 249. Ele declara nível de API 2.0, threadsafety=2 e paramstyle="pyformat". O módulo também fornece os construtores de tipo do PEP 249 Date, Time, Timestamp e Binary, e as funções DateFromTicks, TimeFromTicks e TimestampFromTicks.

from clickhouse_connect import dbapi

connection = dbapi.connect(
    host="localhost",
    username="default",
    password="password",
    database="default",
)
cursor = connection.cursor()

try:
    cursor.execute(
        "SELECT name FROM system.tables "
        "WHERE database = %(database)s ORDER BY name LIMIT 5",
        {"database": "system"},
    )
    print(cursor.description)
    print(cursor.fetchall())
finally:
    cursor.close()
    connection.close()

Cursor.execute e Cursor.executemany aceitam os argumentos nomeados adicionais settings e query_formats. settings passa configurações do ClickHouse. query_formats aplica formatos de leitura por tipo do ClickHouse quando uma instrução retorna linhas, usando o mesmo mapeamento de Client.query. Cursor.execute também aceita o argumento somente nomeado pyformat_encoded. Seu padrão True segue o contrato pyformat da DB-API. O dialeto SQLAlchemy o define como False quando o compilador de instruções emitiu sinais de porcentagem brutos, portanto, normalmente os aplicativos não devem defini-lo. executemany usa o caminho nativo de inserção em massa do driver para instruções INSERT ... VALUES compatíveis com uma sequência materializada de linhas. fetchone, fetchmany e fetchall consomem o resultado materializado atual.

Cursor.description deriva null_ok do tipo de cada coluna de resultado. Tipos não anuláveis retornam False, e tipos anuláveis retornam True, incluindo wrappers Nullable, Variant e Dynamic. None significa que a anulabilidade é desconhecida. Quando uma consulta que começa com SELECT ou WITH, ignorando comentários iniciais, não retorna linhas nem metadados de coluna, o cursor executa uma consulta de metadados LIMIT 0 para preencher description. Se essa consulta de metadados falhar, description permanece vazio.

O ClickHouse não oferece transações tradicionais por meio desta interface HTTP. Connection.commit() e Connection.rollback() são operações sem efeito. As regras de concorrência de ID de sessão ainda se aplicam quando uma conexão é compartilhada.

Os módulos a seguir fornecem helpers públicos adicionais usados por aplicativos cliente.

A versão do pacote instalado é exposta como a string clickhouse_connect.__version__.

Exceções personalizadas, incluindo a hierarquia de exceções da DB-API 2.0, são definidas em clickhouse_connect.driver.exceptions. DatabaseError e OperationalError expõem um atributo numérico code com o código de erro do ClickHouse e um atributo name com o nome simbólico, como UNKNOWN_TABLE, para que os aplicativos possam tomar decisões com base em exc.code em vez de analisar a mensagem. code é definido mesmo quando show_clickhouse_errors está desabilitado, enquanto name requer detalhes do erro (True ou "scrub"). Ambos são None quando indisponíveis, como em erros de transporte. Use show_clickhouse_errors="scrub" quando os usuários finais precisarem ver erros de SQL sem informações de host ou versão do servidor. A configuração também controla mensagens de StreamFailureError no meio do stream e mensagens genéricas de transporte. Ela controla apenas str(exc). Os erros de transporte ainda são anexados como __cause__, e os rastreamentos de pilha podem conter o host, a URL ou o texto de erro da biblioteca original.

As funções e a classe DT64Param no módulo clickhouse_connect.driver.binding podem ser usadas para construir e escapar corretamente consultas em ClickHouse SQL. Da mesma forma, as funções no módulo clickhouse_connect.driver.parser podem ser usadas para analisar nomes de tipos de dados do ClickHouse.

Para mais informações sobre como usar o ClickHouse Connect em aplicações multithread, multiprocesso e assíncronas/orientadas a eventos, consulte Uso avançado (Casos de uso multithread, multiprocesso e assíncronos/orientados a eventos).

Para usar o asyncio de forma nativa, consulte Uso avançado (AsyncClient).

Para obter informações sobre como gerenciar IDs de sessão do ClickHouse em aplicações multithread ou concorrentes, consulte Uso avançado (Gerenciando IDs de sessão do ClickHouse).

Para saber como personalizar o pool de conexões HTTP em aplicações grandes e multithread, consulte Uso avançado (Personalizando o pool de conexões HTTP).