ClickHouse Database handler

clickhouse.py

import logging
from clickhouse_driver import Client, connect, errors
from clickhouse_driver.dbapi.cursor import Cursor

logger = logging.getLogger(__name__)

class ClickHouseDB:
    """
    Class for working with a ClickHouse server
    """

    def __init__(self, name: str, url: str) -> None:
        self.name = name
        self.url = url
        self._client = None
        self._cursor = None

    def client(self) -> Client | errors.Error:
        """
        Returns a ClickHouse client. If a client has already been created,
        the existing client is returned. Otherwise, a new client is created.

        Raises:
            errors.Error: If there is an error creating the client.
        """
        if self._client is None:
            try:
                self._client = Client.from_url(self.url)
            except errors.Error as e:
                logger.error(str(e))
                raise
        return self._client

    def cursor(self) -> Cursor | errors.Error:
        """
        Returns a ClickHouse cursor. If a cursor has already been created,
        the existing cursor is returned. Otherwise, a new cursor is created.

        Raises:
            errors.Error: If there is an error creating the cursor.
        """
        if self._cursor is None:
            try:
                conn = connect(self.url)
                self._cursor = conn.cursor()
            except errors.Error as e:
                logger.error(str(e))
                raise
        return self._cursor