Merge pull request #2 from sn0wfree/v0.1.4

V0.1.4

Author: sn0wfree

ClickSQL/__init__.py

@@ -1,3 +1,8 @@
 # coding=utf-8
-__version__ = '0.1.3'
-__author__ = 'sn0wfree'
+
+
+from ClickSQL.clickhouse.ClickHouse import ClickHouseTableNode
+
+__version__ = '0.1.4'
+__author__ = 'sn0wfree'
+__all__ = ['ClickHouseTableNode']

ClickSQL/clickhouse/ClickHouse.py

@@ -10,14 +10,14 @@
 import requests
 from aiohttp import ClientSession
 
-nest_asyncio.apply()
+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')
 
-SEMAPHORE = 10
+SEMAPHORE = 10  # control async number for whole query list
 
 from ClickSQL.conf.parse_rfc_1738_args import _parse_rfc1738_args
 
@@ -107,14 +107,14 @@ def __init__(self, **db_settings):
         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._session = ClientSession()
         self.max_async_query_once = 5
         self.is_closed = False
 
         self._test_connection_()
 
     @staticmethod
-    def _check_db_settings(db_settings: dict, available_db_type=[node.__name__]):
+    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:
@@ -176,7 +176,7 @@ async def _post(self, url, sql, session):
                 result = await resp.read()
 
         status = resp.status
-        reason = resp.reason
+        # reason = resp.reason
         if status != 200:
             raise ValueError(result)
         return result
@@ -270,7 +270,6 @@ def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
         :param loop:
         :return:
         """
-        ## TODO warning: Unclosed client session
 
         insert_process = list(map(lambda x: x.lower().startswith(available_queries_insert), sql))
         select_process = list(map(lambda x: x.lower().startswith(available_queries_select), sql))
@@ -292,19 +291,24 @@ def execute(self, *sql, convert_to: str = 'dataframe', loop=None, ):
                                       to_df=to_df)
         return result
 
-    def query(self, sql: str):
+    def query(self, *sql: str):
         """
         require to upgrade
         :param sql:
         :return:
         """
-        result = self.execute(sql, convert_to='dataframe', loop=None, )
+        result = self.execute(*sql, convert_to='dataframe', loop=None, )
         return result
 
 
 class ClickHouseTableNode(ClickHouseBaseNode):
-    def __init__(self, conn_str: str):
-        db_settings = _parse_rfc1738_args(conn_str)
+    def __init__(self, conn_str: (str, dict)):
+        if isinstance(conn_str, str):
+            db_settings = _parse_rfc1738_args(conn_str)
+        elif isinstance(conn_str, dict):
+            db_settings = conn_str
+        else:
+            raise ParameterTypeError(f'conn_str must be str or dict but get: {type(conn_str)}')
         super(ClickHouseTableNode, self).__init__(**db_settings)
 
     @property

ClickSQL/clickhouse/ClickHouseExt.py

@@ -1,6 +1,8 @@
 # coding=utf-8
 import pandas as pd
 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,
@@ -198,7 +200,7 @@ def _create_table_from_df(cls, db: str, table: str, df: pd.DataFrame, key_cols:
     def _check_table_exists(cls, obj, db, table):
         ## todo check the table exists
         pass
-if __name__ == '__main__':
-    pass
 
 
+if __name__ == '__main__':
+    pass