Skip to main content

Bunch of decorators to decorate clickhouse-driver's execute method

Project description

clickhouse-driver-decorators

license python coverage

Bunch of decorators to decorate clickhouse-driver's execute method

Available decorators:

  1. add_column_names -- converts each row to dict or namedtuple with column names as keys;
  2. apply_callback -- applies a given callback to each row of result;
  3. convert_string_to_datetime -- converts string datetime to python datetime object in a needed timezone for a given column. String datetime have to be in UTC;
  4. convert_timestamp_to_datetime -- converts timestamp to python datetime object in a needed timezone for a given column.
  5. transform_to_pandas_dataframe -- creates pandas frame based on result from db

Examples

For a closer look, check out the tests

from typing import Generator

from pandas import DataFrame
from clickhouse_driver import Client

from clickhouse_driver_decorators import *


def replace_empty_with_none(row: tuple) -> Generator:
    yield tuple([None if v == '' else v for v in row])


class A:
    def __init__(self, client: Client):
        self.__client = client

    def get_data(self) -> Generator[]:
        return self.__decorated_execute("""
            select
                toUnixTimestamp(now()) as now_ts,
                formatDateTime(now(), '%F %T', 'UTC') as now_str,
                '' as empty
        """)

    @add_column_names()
    @apply_callback(on_row_callable=replace_empty_with_none)
    @convert_timestamp_to_datetime(columns_to_convert={'now_ts': 'Europe/Moscow'})
    @convert_string_to_datetime(date_format='%Y-%m-%d %H:%M:%S', columns_to_convert={'now_str': 'Europe/Moscow'})
    def __decorated_execute(self, query: str):
        return self.__client.execute(query, with_column_types=True)


class B:
    def __init__(self, client: Client):
        self.__client = client

    def get_pandas_df(self) -> DataFrame:
        return self.__decorated_execute("select 'val1' as col1, 'val2' as col2")

    @transform_to_pandas_dataframe()
    def __decorated_execute(self, query: str):
        return self.__client.execute(query, with_column_types=True)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

clickhouse-driver-decorators-0.2.0.tar.gz (4.6 kB view hashes)

Uploaded source

Built Distribution

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page