clickhouse-connect

import clickhouse_connect import os host = os.environ.get('CH_HOST', 'localhost') port = int(os.environ.get('CH_PORT', 8123)) # Use 8443 for TLS/Cloud username = os.environ.get('CH_USER', 'default') password = os.environ.get('CH_PASSWORD', '') database = os.environ.get('CH_DB', 'default') try: client = clickhouse_connect.get_client( host=host, port=port, username=username, password=password, database=database, secure= (port == 8443) # Automatically use TLS for 8443 ) # Test connection client.ping() print(f"Successfully connected to ClickHouse at {host}:{port}") # Create a table client.command( "CREATE TABLE IF NOT EXISTS my_test_table (", " id UInt64,", " name String,", " value Float64", ") ENGINE MergeTree ORDER BY id" ) print("Table 'my_test_table' created or already exists.") # Insert data data_to_insert = [ [1, 'Alpha', 100.1], [2, 'Beta', 200.2], [3, 'Gamma', 300.3] ] client.insert('my_test_table', data_to_insert, column_names=['id', 'name', 'value']) print("Data inserted into 'my_test_table'.") # Query data result = client.query('SELECT * FROM my_test_table ORDER BY id') print("Query Results:") for row in result.result_set: print(row) # Example with Pandas (requires 'pandas' extra) try: import pandas as pd df = client.query_df('SELECT * FROM my_test_table') print("\nPandas DataFrame Results:") print(df) except ImportError: print("\nSkipping Pandas example: 'pandas' not installed. Install with `pip install clickhouse-connect[pandas]`") finally: if 'client' in locals(): client.close() print("Connection closed.") except Exception as e: print(f"An error occurred: {e}")