clickhouse-pool

import os from clickhouse_pool import ChPool # Configure connection details. For production, use environment variables or a config file. host = os.environ.get('CLICKHOUSE_HOST', 'localhost') port = int(os.environ.get('CLICKHOUSE_PORT', 9000)) # Native protocol port user = os.environ.get('CLICKHOUSE_USER', 'default') password = os.environ.get('CLICKHOUSE_PASSWORD', '') database = os.environ.get('CLICKHOUSE_DB', 'default') # Initialize the connection pool # connections_min and connections_max can be adjusted for your workload pool = ChPool( host=host, port=port, user=user, password=password, database=database, connections_min=5, connections_max=10 ) try: with pool.get_client() as client: # Execute a query result = client.execute("SELECT number, 'hello' FROM system.numbers LIMIT 5") print("Query Result:", result) # Example of an insert # client.execute("CREATE TABLE IF NOT EXISTS my_table (id UInt64, value String) ENGINE = Memory") # client.execute("INSERT INTO my_table VALUES", [(1, 'test1'), (2, 'test2')]) # print("Data inserted.") # result_insert = client.execute("SELECT * FROM my_table") # print("Inserted Data:", result_insert) except Exception as e: print(f"An error occurred: {e}") finally: # Always close all connections in the pool once you're done with it pool.cleanup() print("Connection pool cleaned up.")