quality_tools/lib/clickhouse_tools.py

from clickhouse_driver import Client
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ClickhouseUtil:
    @staticmethod
    def connect_to_clickhouse(host, port, user, password, database):
        try:
            client = Client(host=host, port=port, user=user, password=password, database=database)
            return client
        except Exception as e:
            logger.error("Failed to connect to ClickHouse: %s", e)
            raise
    @staticmethod
    def execute_sql(client, query):
        try:
            result = client.execute(query)
            return result
        except Exception as e:
            logger.error("Failed to execute SQL query: %s", e)
            raise