V0.1.4 (#3)

* alter some code structure and remove useless code

* alter some code structure and remove useless code

Author: sn0wfree

ClickSQL/clickhouse/ClickHouse.py

@@ -4,36 +4,52 @@
 import json
 from collections import namedtuple
 from urllib import parse
-
-import nest_asyncio
 import pandas as pd
+import nest_asyncio
 import requests
 from aiohttp import ClientSession
+from ClickSQL.conf.parse_rfc_1738_args import parse_rfc1738_args
+
+"""
+this will hold base function of clickhouse and it will apply a path of access clickhouse through clickhouse api service
+this scripts will use none of clickhouse client and only depend on requests to make transactions with 
+clickhouse-server
+
+"""
 
 nest_asyncio.apply()  # allow run at jupyter and asyncio env
 
 node_parameters = ['host', 'port', 'user', 'password', 'database']
 node = namedtuple('clickhouse', node_parameters)
 available_queries_select = ('select', 'show', 'desc')
 available_queries_insert = ('insert', 'optimize', 'create')
-
+PRINT_TEST_RESULT = True
 SEMAPHORE = 10  # control async number for whole query list
 
-from ClickSQL.conf.parse_rfc_1738_args import _parse_rfc1738_args
 
-
-class ParameterKeyError(Exception): pass
+class ParameterKeyError(Exception):
+    pass
 
 
-class ParameterTypeError(Exception): pass
+class ParameterTypeError(Exception):
+    pass
 
 
-class DatabaseTypeError(Exception): pass
+class DatabaseTypeError(Exception):
+    pass
 
 
+# TODO change to queue mode change remove aiohttp depends
 class ClickHouseTools(object):
     @staticmethod
     def _transfer_sql_format(sql, convert_to, transfer_sql_format=True):
+        """
+        provide a method which will translate a standard sql into clickhouse sql with might use format as suffix
+        :param sql:
+        :param convert_to:
+        :param transfer_sql_format:
+        :return:
+        """
         if transfer_sql_format:
             clickhouse_format = 'JSON' if convert_to is None else 'JSONCompact' if convert_to.lower() == 'dataframe' else convert_to
             query_with_format = (sql.rstrip('; \n\t') + ' format ' + clickhouse_format).replace('\n', ' ').strip(' ')
@@ -43,78 +59,100 @@ def _transfer_sql_format(sql, convert_to, transfer_sql_format=True):
 
     @staticmethod
     def _load_into_pd(ret_value, convert_to: str = 'dataframe', errors='ignore'):
+        """
+        will provide a approach to load data from clickhouse into pd.DataFrame format which may be easy to use
+
+        :param ret_value:
+        :param convert_to:
+        :param errors:
+        :return:
+        """
+
         if convert_to.lower() == 'dataframe':
+
             result_dict = json.loads(ret_value, strict=False)
             meta = result_dict['meta']
             name = map(lambda x: x['name'], meta)
             data = result_dict['data']
             df = pd.DataFrame.from_records(data, columns=list(name))
 
             for i in meta:
-                if i['type'] in ['DateTime', 'Nullable(DateTime)']:
+                if i['type'] in ['DateTime', 'Nullable(DateTime)']:  # translate format
+                    # process datetime format
                     df[i['name']] = pd.to_datetime(df[i['name']], errors=errors)
-            ret_value = df
-        return ret_value
+            return df
+        else:
+            return ret_value
 
     @classmethod
-    def _merge_settings(cls, settings, updated_settings=None):
+    def _merge_settings(cls, settings: (None, dict), updated_settings: (None, dict) = None,
+                        extra_settings: (None, dict) = None):
         """
 
         :param settings:
         :param updated_settings:
         :return:
         """
+
         if updated_settings is None:
             updated_settings = {'enable_http_compression': 1, 'send_progress_in_http_headers': 0,
                                 'log_queries': 1, 'connect_timeout': 10, 'receive_timeout': 300,
                                 'send_timeout': 300, 'output_format_json_quote_64bit_integers': 0,
                                 'wait_end_of_query': 0}
+        elif not isinstance(updated_settings, dict):
+            raise ParameterTypeError(f'updated_settings must be dict type, but get {type(updated_settings)}')
+        else:
+            pass
 
-        if settings is not None:
+        if settings is not None and isinstance(settings, dict):
             invalid_setting_keys = list(set(settings.keys()) - set(updated_settings.keys()))
             if len(invalid_setting_keys) > 0:
                 raise ValueError('setting "{0}" is invalid, valid settings are: {1}'.format(
                     invalid_setting_keys[0], ', '.join(updated_settings.keys())))
-            else:
-                pass
+
             updated_settings.update(settings)
 
+        if extra_settings is not None and isinstance(extra_settings, dict):
+            updated_settings.update(extra_settings)
+
         return {k: v * 1 if isinstance(v, bool) else v for k, v in updated_settings.items()}
 
 
 class ClickHouseBaseNode(ClickHouseTools):
-    accepted_formats = ['DataFrame', 'TabSeparated', 'TabSeparatedRaw', 'TabSeparatedWithNames',
+    accepted_formats = ('DataFrame', 'TabSeparated', 'TabSeparatedRaw', 'TabSeparatedWithNames',
                         'TabSeparatedWithNamesAndTypes', 'CSV', 'CSVWithNames', 'Values', 'Vertical', 'JSON',
                         'JSONCompact', 'JSONEachRow', 'TSKV', 'Pretty', 'PrettyCompact',
-                        'PrettyCompactMonoBlock', 'PrettyNoEscapes', 'PrettySpace', 'XML']
+                        'PrettyCompactMonoBlock', 'PrettyNoEscapes', 'PrettySpace', 'XML')
+
     _default_settings = {'enable_http_compression': 1, 'send_progress_in_http_headers': 0,
                          'log_queries': 1, 'connect_timeout': 10, 'receive_timeout': 300,
                          'send_timeout': 300, 'output_format_json_quote_64bit_integers': 0,
                          'wait_end_of_query': 0}
 
+    __slots__ = ('_db', '_base_url', '_para', 'http_settings', 'max_async_query_once', 'is_closed')
+
     def __init__(self, **db_settings):
         """
         :param db_settings:
         """
-        self._check_db_settings(db_settings)
+        self._check_db_settings(db_settings, available_db_type=[node.__name__])
 
+        self._db = db_settings['database']
+        self._base_url = "http://{host}:{port}/?".format(host=db_settings['host'], port=int(db_settings['port']))
         self._para = node(db_settings['host'], db_settings['port'], db_settings['user'],
-                          db_settings['password'], db_settings['database'])
-        self._db = self._para.database
-
-        self._base_url = "http://{host}:{port}/?".format(host=self._para.host, port=int(self._para.port))
+                          db_settings['password'], db_settings['database'])  # store connection information
 
-        self.http_settings = self._merge_settings(None, updated_settings=self._default_settings)
-        self.http_settings.update({'user': self._para.user, 'password': self._para.password})
-
-        # self._session = ClientSession()
+        self.http_settings = self._merge_settings(None, updated_settings=self._default_settings,
+                                                  extra_settings={'user': self._para.user,
+                                                                  'password': self._para.password})
+        # self._session = ClientSession() # the reason of unclose session client
         self.max_async_query_once = 5
         self.is_closed = False
 
-        self._test_connection_()
+        self._test_connection_(self._base_url)
 
     @staticmethod
-    def _check_db_settings(db_settings: dict, available_db_type=[node.__name__]):  # node.__name__ : clickhouse
+    def _check_db_settings(db_settings: dict, available_db_type=(node.__name__,)):  # node.__name__ : clickhouse
         """
         it is to check db setting whether is correct!
         :param db_settings:
@@ -123,27 +161,32 @@ def _check_db_settings(db_settings: dict, available_db_type=[node.__name__]):  #
 
         if isinstance(db_settings, dict):
             if db_settings['name'].lower() not in available_db_type:
-                raise DatabaseTypeError('database symbol is not accept, now only accept: {",".join(available_db_type)}')
-            missing_keys = []
-            for key in node_parameters:
-                if key not in db_settings.keys():
-                    missing_keys.append(key)
-                else:
-                    pass
+                raise DatabaseTypeError(
+                    f'database symbol is not accept, now only accept: {",".join(available_db_type)}')
+            missing_keys = [key for key in node_parameters if key not in db_settings.keys()]
+            # :
+            #     missing_keys.append(key)
+            # else:
+            #     pass
             if len(missing_keys) == 0:
                 pass
             else:
                 raise ParameterKeyError(f"the following keys are not at settings: {','.join(missing_keys)}")
         else:
             raise ParameterTypeError(f'db_setting must be dict type! but get {type(db_settings)}')
 
-    def _test_connection_(self):
+    @staticmethod
+    def _test_connection_(_base_url):
         """
         is to test connection by normal way!
+
+        alter function type into staticmethod
         :return:
         """
-        ret_value = requests.get(self._base_url)
-        print('connection test: ', ret_value.text.strip())
+        ret_value = requests.get(_base_url)
+        if PRINT_TEST_RESULT:
+            print('connection test: ', ret_value.text.strip())
+        del ret_value
 
     @property
     def _connect_url(self):
@@ -158,9 +201,9 @@ def _connect_url(self):
                                                                 )
         return url_str
 
-    async def _post(self, url, sql, session):
+    async def _post(self, url: str, sql: str, session):
         """
-        the aysnc way to send post request to the server
+        the async way to send post request to the server
         :param url:
         :param sql:
         :param session:
@@ -213,27 +256,28 @@ async def _compression_switched_request(self, query_with_format: str, convert_to
 
         return result
 
-    def _load_into_pd_ext(self, sql, res, convert_to, to_df):
+    @classmethod
+    def _load_into_pd_ext(cls, sql: (str, list, tuple), ret_value, convert_to: str, to_df: bool):
         """
         a way to parse into dataframe
         :param sql:
-        :param res:
+        :param ret_value:
         :param convert_to:
         :param to_df:
         :return:
         """
         if isinstance(sql, str):
-            if to_df:
-                result = self._load_into_pd(res, convert_to)
+            if to_df or ret_value != b'':
+                result = cls._load_into_pd(ret_value, convert_to)
             else:
-                result = res
+                result = ret_value
         elif isinstance(sql, (list, tuple)):
             if to_df:
-                result = [self._load_into_pd(s, convert_to) for s in res]
+                result = [cls._load_into_pd(s, convert_to) if ret_value != b'' else None for s in ret_value]
             else:
-                result = res
+                result = ret_value
         else:
-            raise ValueError('sql must be str or list or tuple')
+            raise ValueError(f'sql must be str or list or tuple,but get {type(sql)}')
         return result
 
     def __execute__(self, sql: (str, list, tuple), convert_to: str = 'dataframe', transfer_sql_format: bool = True,
@@ -252,26 +296,26 @@ def __execute__(self, sql: (str, list, tuple), convert_to: str = 'dataframe', tr
             loop = asyncio.get_event_loop()
 
         sem = asyncio.Semaphore(SEMAPHORE)  # limit async num
-
         resp_list = self._compression_switched_request(sql, convert_to=convert_to,
                                                        transfer_sql_format=transfer_sql_format, sem=sem)
 
         res = loop.run_until_complete(resp_list)
-
         result = self._load_into_pd_ext(sql, res, convert_to, to_df)
 
         return result
 
     def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
         """
-
+        execute sql or multi sql
         :param sql:
         :param convert_to:
         :param loop:
         :return:
         """
-
+        # TODO change to smart mode, can receive any kind sql combination and handle them
+        # detect whether all query are insert process
         insert_process = list(map(lambda x: x.lower().startswith(available_queries_insert), sql))
+        # detect whether all query are select process
         select_process = list(map(lambda x: x.lower().startswith(available_queries_select), sql))
         if all(insert_process) is True:
             to_df = False
@@ -280,10 +324,12 @@ def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
             to_df = True
             transfer_sql_format = True
         else:
-            raise ValueError('the list of query must be same type query!')
+            # TODO change to smart mode, can receive any kind sql combination and handle them
+            raise ValueError(
+                'the list of queries must be same type query! currently cannot handle various kind SQL type'
+                'combination')
 
         if len(sql) != 1:
-
             result = self.__execute__(sql, convert_to=convert_to, transfer_sql_format=transfer_sql_format, loop=loop,
                                       to_df=to_df)
         else:
@@ -293,7 +339,7 @@ def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
 
     def query(self, *sql: str):
         """
-        require to upgrade
+        ## TODO require to upgrade
         :param sql:
         :return:
         """
@@ -304,7 +350,7 @@ def query(self, *sql: str):
 class ClickHouseTableNode(ClickHouseBaseNode):
     def __init__(self, conn_str: (str, dict)):
         if isinstance(conn_str, str):
-            db_settings = _parse_rfc1738_args(conn_str)
+            db_settings = parse_rfc1738_args(conn_str)
         elif isinstance(conn_str, dict):
             db_settings = conn_str
         else:

ClickSQL/clickhouse/ClickHouseExt.py

@@ -3,6 +3,7 @@
 import numpy as np
 import re
 
+
 class ClickHouseTableNodeExt(object):
     @classmethod
     def _create_table(cls, obj: object, db: str, table: str, sql: str, key_cols: list,