clickhouse-sqlalchemy

import os from sqlalchemy import create_engine, text CH_HOST = os.environ.get('CLICKHOUSE_HOST', 'localhost') CH_PORT = os.environ.get('CLICKHOUSE_PORT', '8123') CH_USER = os.environ.get('CLICKHOUSE_USER', 'default') CH_PASSWORD = os.environ.get('CLICKHOUSE_PASSWORD', '') CH_DATABASE = os.environ.get('CLICKHOUSE_DATABASE', 'default') try: # Establish a connection to ClickHouse engine = create_engine( f'clickhouse://{CH_USER}:{CH_PASSWORD}@{CH_HOST}:{CH_PORT}/{CH_DATABASE}' ) # Execute a simple query with engine.connect() as connection: result = connection.execute(text('SELECT 1 as one')).scalar() print(f"Query result: {result}") # Example: Create a table and insert data with engine.connect() as connection: connection.execute(text('DROP TABLE IF EXISTS my_test_table')) connection.execute(text('CREATE TABLE my_test_table (id Int32, name String) ENGINE = MergeTree ORDER BY id')) connection.execute(text('INSERT INTO my_test_table (id, name) VALUES (1, \'Alice\'), (2, \'Bob\')')) connection.commit() # Query data rows = connection.execute(text('SELECT id, name FROM my_test_table ORDER BY id')).fetchall() print(f"Table data: {rows}") except Exception as e: print(f"An error occurred: {e}") print("Ensure a ClickHouse instance is running and connection details are correct.") print("You can set environment variables like CLICKHOUSE_HOST, CLICKHOUSE_PORT, etc.")