Merge branch 'v0.1.1'

Author: sn0wfree

.gitignore

@@ -127,3 +127,5 @@ dmypy.json
 .idea
 # Pyre type checker
 .pyre/
+
+.pypirc

ClickSQL/__init__.py

@@ -1,2 +1,3 @@
-__version__ = '0.0.1'
+# coding=utf-8
+__version__ = '0.1.2'
 __author__ = 'sn0wfree'

ClickSQL/conf/__init__.py

@@ -1,8 +1,5 @@
 # coding=utf-8
-import pandas as pd
-import numpy as np
+
 
 if __name__ == '__main__':
     pass
-
-

ClickSQL/core/__init__.py

@@ -1,8 +1,5 @@
 # coding=utf-8
-import pandas as pd
-import numpy as np
+
 
 if __name__ == '__main__':
     pass
-
-

ClickSQL/core/clickhouse.py

@@ -6,40 +6,133 @@
 from collections import namedtuple
 from urllib import parse
 
+import nest_asyncio
 import pandas as pd
 import requests
 from aiohttp import ClientSession
 
+nest_asyncio.apply()
 node = namedtuple('clickhouse', ['host', 'port', 'user', 'password', 'database'])
 available_queries_select = ('select', 'show', 'desc')
 available_queries_insert = ('insert', 'optimize', 'create')
+
 SEMAPHORE = 10
 
 
-class ClickhouseTools(object):
+class ClickHouseCreateTableTools(object):
+    @classmethod
+    def _create_table_sql(cls, db: str, table: str, dtypes_dict: dict,
+                          order_by_key_cols: (list, tuple),
+                          primary_key_cols=None, sample_expr=None,
+                          engine_type: str = 'ReplacingMergeTree', partitions_expr=None,
+                          settings="SETTINGS index_granularity = 8192", other=''):
+        """
+
+        :param db:
+        :param table:
+        :param dtypes_dict:
+        :param order_by_key_cols:
+        :param primary_key_cols:
+        :param sample_expr:
+        :param engine_type:
+        :param partitions_expr:
+        :param settings:
+        :param other:
+        :return:
+        """
+        """CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
+                   (
+                       name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
+                       name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
+                       ...
+                   ) ENGINE = ReplacingMergeTree([ver])
+                   [PARTITION BY expr]
+                   [ORDER BY expr]
+                   [PRIMARY KEY expr]
+                   [SAMPLE BY expr]
+                   [SETTINGS name=value, ...]"""
+
+        cols_def = ','.join([f"{name} {d_type}" for name, d_type in dtypes_dict.items()])
+
+        maid_body = f"CREATE TABLE IF NOT EXISTS {db}.{table} ( {cols_def} ) ENGINE = {engine_type}"
+
+        ORDER_BY_CLAUSE = f"ORDER BY ( {','.join(order_by_key_cols)} )"
+
+        if partitions_expr is None:
+            PARTITION_by_CLAUSE = ''
+        else:
+            PARTITION_by_CLAUSE = f"PARTITION BY {partitions_expr}"
+
+        if primary_key_cols is None:
+            PRIMARY_BY_CLAUSE = ''
+        else:
+            primary_key_expr = ','.join(primary_key_cols)
+            PRIMARY_BY_CLAUSE = f"PARTITION BY ({primary_key_expr})"
+
+        if sample_expr is None:
+            SAMPLE_CLAUSE = ''
+        else:
+            SAMPLE_CLAUSE = sample_expr
+
+        base = f"{maid_body} {PARTITION_by_CLAUSE} {ORDER_BY_CLAUSE} {PRIMARY_BY_CLAUSE} {SAMPLE_CLAUSE} {other} {settings}"
+
+        return base
+
     @staticmethod
-    def detect_end_with_limit(string, pattern=r'[\s]+limit[\s]+[0-9]+$'):
+    def _check_end_with_limit(string, pattern=r'[\s]+limit[\s]+[0-9]+$'):
         m = re.findall(pattern, string)
         if m is None or m == []:
             return False
         else:
             return True
 
-    @staticmethod
-    def translate_dtypes_from_df(df: pd.DataFrame,
-                                 translate_dtypes: dict = {'object': 'String',
-                                                           'datetime64[ns]': 'Datetime'}):
-        if hasattr(df, 'dtypes'):
-            dtypes_series = df.dtypes.replace(translate_dtypes)
-            return dtypes_series.map(lambda x: str(x).capitalize()).to_dict()
-        elif hasattr(df, '_columns_') and 'type' in df._columns_ and 'name' in df._columns_:
-            dtypes_series = df.set_index('name')['type'].replace(translate_dtypes)
-            return dtypes_series.map(lambda x: str(x)).to_dict()
+    @classmethod
+    def _create_table_from_sql(cls, db: str, table: str, sql: str, key_cols: list,
+                               extra_format_dict: (dict, None) = None,
+                               primary_key_cols=None, sample_expr=None, other='',
+                               engine_type: str = 'ReplacingMergeTree',
+                               partitions_expr=None, query_func=None):
+        """
+
+        :param obj:
+        :param db:
+        :param table:
+        :param sql:
+        :param key_cols:
+        :param engine_type:
+        :param extra_format_dict:
+        :return:
+        """
+
+        if isinstance(sql, str):
+            pass
         else:
-            raise ValueError(f'unknown df:{type(df)}')
+            raise ValueError('sql must be string')
+
+        limit_status = cls._check_end_with_limit(sql, pattern=r'[\s]+limit[\s]+[0-9]+$')
+        if limit_status:
+            describe_sql = f' describe({sql}) '
+        else:
+            describe_sql = f'describe ( {sql} limit 1)'
+
+        if query_func is None:
+            raise ValueError('query function should be set!')
+
+        dtypes_df = query_func(describe_sql)
+
+        dtypes_dict = dict(dtypes_df[['name', 'type']].drop_duplicates().values)
+        if extra_format_dict is None:
+            pass
+        else:
+            dtypes_dict.update(extra_format_dict)
+        sql = cls._create_table_sql(db, table, dtypes_dict, key_cols, engine_type=engine_type,
+                                    primary_key_cols=primary_key_cols, sample_expr=sample_expr,
+                                    partitions_expr=partitions_expr, other=other)
+
+        return sql
 
     @staticmethod
-    def translate_dtype1_as_dtype2(df: pd.DataFrame, src2target={'category': 'str'}):
+    def translate_dtypes1_as_dtypes2(df: pd.DataFrame, src2target={'category': 'str'}):
         dtypes_series = df.dtypes
         for src, dest in src2target.items():
             if src in dtypes_series:
@@ -50,12 +143,26 @@ def translate_dtype1_as_dtype2(df: pd.DataFrame, src2target={'category': 'str'})
                 pass
         return df
 
+    @staticmethod
+    def translate_dtypes_from_df(df: pd.DataFrame, translate_dtypes: dict = {'object': 'String',
+                                                                             'datetime64[ns]': 'Datetime'}):
+        if hasattr(df, 'dtypes'):
+            dtypes_series = df.dtypes.replace(translate_dtypes)
+            return dtypes_series.map(lambda x: str(x).capitalize()).to_dict()
+        elif hasattr(df, '_columns_') and 'type' in df._columns_ and 'name' in df._columns_:
+            dtypes_series = df.set_index('name')['type'].replace(translate_dtypes)
+            return dtypes_series.map(lambda x: str(x)).to_dict()
+        else:
+            raise ValueError(f'unknown df:{type(df)}')
+
     @classmethod
-    def _create_table_from_df(cls, obj: object, db: str, table: str, df: pd.DataFrame, key_cols: (list, tuple),
-                              engine_type: str = 'ReplacingMergeTree', extra_format_dict=None, partitions_expr=None):
-        query_func = obj.query
+    def _create_table_from_df(cls, db: str, table: str, df: pd.DataFrame, key_cols: (list, tuple),
+                              engine_type: str = 'ReplacingMergeTree', extra_format_dict=None, partitions_expr=None,
+                              src2target={'category': 'str'},
+                              query_func=None
+                              ):
 
-        df = cls.translate_dtype1_as_dtype2(df, src2target={'category': 'str'})
+        df = cls.translate_dtypes1_as_dtypes2(df, src2target={'category': 'str'})
         cols = df.columns
         dtypes_dict = cls.translate_dtypes_from_df(df)
         if extra_format_dict is None:
@@ -71,65 +178,31 @@ def _create_table_from_df(cls, obj: object, db: str, table: str, df: pd.DataFram
         return exist_status
 
     @classmethod
-    def _create_table_sql(cls, db: str, table: str, dtypes_dict: dict, key_cols: (list, tuple),
-                          engine_type: str = 'ReplacingMergeTree', partitions_expr=None):
-        # dtypes_dict.update(extra_format_dict)
-        cols_def = ','.join([f"{name} {d_type}" for name, d_type in dtypes_dict.items()])
-        order_by_cols = ','.join(key_cols)
+    def _check_table_exists(cls, obj, db, table):
+        ## todo check the table exists
+        pass
 
-        maid_body = f"CREATE TABLE IF NOT EXISTS {db}.{table} ( {cols_def} ) ENGINE = {engine_type}"
-        settings = "SETTINGS index_granularity = 8192"
-        conds = f"ORDER BY ( {order_by_cols} )"
-        if partitions_expr is None:
-            partitions = ''
-        else:
-            partitions = f"PARTITION BY {partitions_expr}"
-        base = f"{maid_body} {conds} {partitions} {settings}"
-        return base
 
-    @classmethod
-    def _create_table_from_sql(cls, obj: object, db: str, table: str, sql: str, key_cols: list,
-                               engine_type: str = 'ReplacingMergeTree',
-                               extra_format_dict: (dict, None) = None,
-                               partitions_expr=None) -> bool:
-        """
+class ClickHouseTools(ClickHouseCreateTableTools):
 
-        :param obj:
-        :param db:
-        :param table:
-        :param sql:
-        :param key_cols:
-        :param engine_type:
-        :param extra_format_dict:
-        :return:
-        """
+    @classmethod
+    def _create_table_from_df(cls, obj: object, db: str, table: str, df: pd.DataFrame, key_cols: (list, tuple),
+                              engine_type: str = 'ReplacingMergeTree', extra_format_dict=None, partitions_expr=None):
+        query_func = obj.query
 
+        df = cls.translate_dtypes1_as_dtypes2(df, src2target={'category': 'str'})
+        cols = df.columns
+        dtypes_dict = cls.translate_dtypes_from_df(df)
         if extra_format_dict is None:
-            extra_format_dict = {}
-
-        query_func = obj.query
-        if sql.endswith(';'):
-            sql = sql[:-1]
-        end_with_limit_status = cls.detect_end_with_limit(sql, pattern=r'[\s]+limit[\s]+[0-9]+$')
-        if end_with_limit_status:
-            describe_sql = f' describe({sql}) '
+            pass
         else:
-            describe_sql = f'describe ( {sql} limit 1)'
-
+            dtypes_dict.update(extra_format_dict)
+        dtypes_dict = {k: v for k, v in dtypes_dict.items() if k in cols}
+        base = cls._create_table_from_sql(db, table, dtypes_dict, key_cols, engine_type=engine_type,
+                                          extra_format_dict=extra_format_dict, partitions_expr=partitions_expr)
         exist_status = cls._check_table_exists(obj, db, table)
-        if exist_status:
-            print('table:{table} already exists!')
-        else:
-            print('will create {table} at {db}')
-            dtypes_df = query_func(describe_sql)
-            dtypes_dict = dict(dtypes_df[['name', 'type']].drop_duplicates().values)
-            if extra_format_dict is None:
-                pass
-            else:
-                dtypes_dict.update(extra_format_dict)
-            sql = cls._create_table_sql(db, table, dtypes_dict, key_cols, engine_type=engine_type,
-                                        partitions_expr=partitions_expr)
-            query_func(sql)
+
+        query_func(base)
         return exist_status
 
     @classmethod
@@ -167,7 +240,7 @@ def _load_into_pd(ret_value, convert_to: str = 'dataframe'):
 
             for i in meta:
                 if i['type'] in ['DateTime', 'Nullable(DateTime)']:
-                    df[i['name']] = pd.to_datetime(df[i['name']])
+                    df[i['name']] = pd.to_datetime(df[i['name']], errors='ignore')
             ret_value = df
         return ret_value
 
@@ -197,7 +270,7 @@ def _merge_settings(cls, settings, updated_settings=None):
         return {k: v * 1 if isinstance(v, bool) else v for k, v in updated_settings.items()}
 
 
-class ClickhouseBaseNode(ClickhouseTools):
+class ClickhouseBaseNode(ClickHouseTools):
     accepted_formats = ['DataFrame', 'TabSeparated', 'TabSeparatedRaw', 'TabSeparatedWithNames',
                         'TabSeparatedWithNamesAndTypes', 'CSV', 'CSVWithNames', 'Values', 'Vertical', 'JSON',
                         'JSONCompact', 'JSONEachRow', 'TSKV', 'Pretty', 'PrettyCompact',
@@ -267,21 +340,28 @@ async def _compression_switched_request(self, query_with_format: str, convert_to
                 if isinstance(query_with_format, str):
                     sql2 = self._transfer_sql_format(query_with_format, convert_to=convert_to,
                                                      transfer_sql_format=transfer_sql_format)
-
                     result = await self._post(url, sql2, session)
-
                 elif isinstance(query_with_format, (tuple, list)):
                     result = []
                     for sql in query_with_format:
                         s = self._transfer_sql_format(sql, convert_to=convert_to,
                                                       transfer_sql_format=transfer_sql_format)
-
                         res = await self._post(url, s, session)
                         result.append(res)
-
                 else:
-                    raise ValueError('query_with_format must be str or list or tuple')
+                    raise ValueError('query_with_format must be str , list or tuple')
+
+        return result
 
+    def _load_into_pd_ext(self, sql, res, convert_to, to_df):
+        if isinstance(sql, str):
+            if to_df:
+                result = self._load_into_pd(res, convert_to)
+        elif isinstance(sql, (list, tuple)):
+            if to_df:
+                result = [self._load_into_pd(s, convert_to) for s in res]
+        else:
+            raise ValueError('sql must be str or list or tuple')
         return result
 
     def __execute__(self, sql: (str, list, tuple), convert_to: str = 'dataframe', transfer_sql_format: bool = True,
@@ -306,14 +386,7 @@ def __execute__(self, sql: (str, list, tuple), convert_to: str = 'dataframe', tr
 
         res = loop.run_until_complete(resp_list)
 
-        if isinstance(sql, str):
-            if to_df:
-                result = self._load_into_pd(res, convert_to)
-        elif isinstance(sql, (list, tuple)):
-            if to_df:
-                result = [self._load_into_pd(s, convert_to) for s in res]
-        else:
-            raise ValueError('sql must be str or list or tuple')
+        result = self._load_into_pd_ext(sql, res, convert_to, to_df)
 
         return result
 
@@ -336,7 +409,7 @@ def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
             to_df = True
             transfer_sql_format = True
         else:
-            raise ValueError('sql list must be same type query!')
+            raise ValueError('the list of query must be same type query!')
 
         if len(sql) != 1:
 
@@ -376,5 +449,5 @@ def databases(self):
         **{'host': '47.104.186.157', 'port': 8123, 'user': 'default', 'password': 'Imsn0wfree',
            'database': 'EDGAR_LOG'})
     df1 = node.tables
-    print(df1)
+    dff = node.query("select * from system.parts")
     pass

ClickSQL/utils/__init__.py

@@ -1,8 +1,5 @@
 # coding=utf-8
-import pandas as pd
-import numpy as np
+
 
 if __name__ == '__main__':
     pass
-
-