import clickhouse_connect
def connect_to_clickhouse():
# Replace with your ClickHouse host, port, username, and password
host = 'connector-hostname'
port = 8444
username = 'idp:formal:human:john@joinformal.com'
password = 'your-formal-password'
try:
# Initialize the ClickHouse client
client = clickhouse_connect.get_client(
host=host,
user=username,
password=password,
port=8444,
database='default',
)
print("Connection to ClickHouse established successfully!")
return client
except Exception as e:
print(f"Failed to connect to ClickHouse: {e}")
return None
def main():
client = connect_to_clickhouse()
if client:
# Example query to check connection
try:
print("Result:", client.query("SELECT 1").result_set[0][0])
except Exception as e:
print(f"Error executing query: {e}")
if __name__ == "__main__":
main()