From ab7cd4faab19fdf2f0929cda9060bbc97ecb699c Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 25 Jan 2021 15:35:49 +0800 Subject: [PATCH] [TD-2843] feature: porting python connector for mac. --- src/connector/python/osx/python3/LICENSE | 12 + src/connector/python/osx/python3/README.md | 1 + src/connector/python/osx/python3/setup.py | 20 + .../python/osx/python3/taos/__init__.py | 24 + .../python/osx/python3/taos/cinterface.py | 477 ++++++++++++++++++ .../python/osx/python3/taos/connection.py | 86 ++++ .../python/osx/python3/taos/constants.py | 33 ++ .../python/osx/python3/taos/cursor.py | 271 ++++++++++ .../python/osx/python3/taos/dbapi.py | 38 ++ .../python/osx/python3/taos/error.py | 57 +++ .../python/osx/python3/taos/subscription.py | 52 ++ 11 files changed, 1071 insertions(+) create mode 100644 src/connector/python/osx/python3/LICENSE create mode 100644 src/connector/python/osx/python3/README.md create mode 100644 src/connector/python/osx/python3/setup.py create mode 100644 src/connector/python/osx/python3/taos/__init__.py create mode 100644 src/connector/python/osx/python3/taos/cinterface.py create mode 100644 src/connector/python/osx/python3/taos/connection.py create mode 100644 src/connector/python/osx/python3/taos/constants.py create mode 100644 src/connector/python/osx/python3/taos/cursor.py create mode 100644 src/connector/python/osx/python3/taos/dbapi.py create mode 100644 src/connector/python/osx/python3/taos/error.py create mode 100644 src/connector/python/osx/python3/taos/subscription.py diff --git a/src/connector/python/osx/python3/LICENSE b/src/connector/python/osx/python3/LICENSE new file mode 100644 index 0000000000..79a9d73086 --- /dev/null +++ b/src/connector/python/osx/python3/LICENSE @@ -0,0 +1,12 @@ + Copyright (c) 2019 TAOS Data, Inc. + +This program is free software: you can use, redistribute, and/or modify +it under the terms of the GNU Affero General Public License, version 3 +or later ("AGPL"), as published by the Free Software Foundation. + +This program is distributed in the hope that it will be useful, but WITHOUT +ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +FITNESS FOR A PARTICULAR PURPOSE. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . diff --git a/src/connector/python/osx/python3/README.md b/src/connector/python/osx/python3/README.md new file mode 100644 index 0000000000..70db6bba13 --- /dev/null +++ b/src/connector/python/osx/python3/README.md @@ -0,0 +1 @@ +# TDengine python client interface \ No newline at end of file diff --git a/src/connector/python/osx/python3/setup.py b/src/connector/python/osx/python3/setup.py new file mode 100644 index 0000000000..098f786d62 --- /dev/null +++ b/src/connector/python/osx/python3/setup.py @@ -0,0 +1,20 @@ +import setuptools + +with open("README.md", "r") as fh: + long_description = fh.read() + +setuptools.setup( + name="taos", + version="2.0.4", + author="Taosdata Inc.", + author_email="support@taosdata.com", + description="TDengine python client package", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://github.com/pypa/sampleproject", + packages=setuptools.find_packages(), + classifiers=[ + "Programming Language :: Python :: 3", + "Operating System :: MacOS X", + ], +) diff --git a/src/connector/python/osx/python3/taos/__init__.py b/src/connector/python/osx/python3/taos/__init__.py new file mode 100644 index 0000000000..216214e097 --- /dev/null +++ b/src/connector/python/osx/python3/taos/__init__.py @@ -0,0 +1,24 @@ + +from .connection import TDengineConnection +from .cursor import TDengineCursor + +# Globals +apilevel = '2.0.4' +threadsafety = 0 +paramstyle = 'pyformat' + +__all__ = ['connection', 'cursor'] + +def connect(*args, **kwargs): + """ Function to return a TDengine connector object + + Current supporting keyword parameters: + @dsn: Data source name as string + @user: Username as string(optional) + @password: Password as string(optional) + @host: Hostname(optional) + @database: Database name(optional) + + @rtype: TDengineConnector + """ + return TDengineConnection(*args, **kwargs) diff --git a/src/connector/python/osx/python3/taos/cinterface.py b/src/connector/python/osx/python3/taos/cinterface.py new file mode 100644 index 0000000000..2cd54d536b --- /dev/null +++ b/src/connector/python/osx/python3/taos/cinterface.py @@ -0,0 +1,477 @@ +import ctypes +from .constants import FieldType +from .error import * +import math +import datetime + +def _convert_millisecond_to_datetime(milli): + return datetime.datetime.fromtimestamp(milli/1000.0) + +def _convert_microsecond_to_datetime(micro): + return datetime.datetime.fromtimestamp(micro/1000000.0) + +def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bool row to python row + """ + _timestamp_converter = _convert_millisecond_to_datetime + if micro: + _timestamp_converter = _convert_microsecond_to_datetime + + if num_of_rows > 0: + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)])) + else: + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)])) + +def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bool row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)] ] + else: + return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[:abs(num_of_rows)] ] + +def _crow_tinyint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C tinyint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)] ] + else: + return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)] ] + +def _crow_smallint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C smallint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)]] + else: + return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)] ] + +def _crow_int_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C int row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)] ] + else: + return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)] ] + +def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bigint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)] ] + else: + return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)] ] + +def _crow_float_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C float row to python row + """ + if num_of_rows > 0: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)] ] + else: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)] ] + +def _crow_double_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C double row to python row + """ + if num_of_rows > 0: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)] ] + else: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)] ] + +def _crow_binary_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C binary row to python row + """ + assert(nbytes is not None) + if num_of_rows > 0: + return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)]] + else: + return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)]] + +def _crow_nchar_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C nchar row to python row + """ + assert(nbytes is not None) + res=[] + for i in range(abs(num_of_rows)): + try: + if num_of_rows >= 0: + tmpstr = ctypes.c_char_p(data) + res.append( tmpstr.value.decode() ) + else: + res.append( (ctypes.cast(data+nbytes*i, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) + except ValueError: + res.append(None) + + return res + +def _crow_binary_to_python_block(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C binary row to python row + """ + assert(nbytes is not None) + res=[] + if num_of_rows > 0: + for i in range(abs(num_of_rows)): + try: + rbyte=ctypes.cast(data+nbytes*i,ctypes.POINTER(ctypes.c_short))[:1].pop() + tmpstr = ctypes.c_char_p(data+nbytes*i+2) + res.append( tmpstr.value.decode()[0:rbyte] ) + except ValueError: + res.append(None) + else: + for i in range(abs(num_of_rows)): + try: + rbyte=ctypes.cast(data+nbytes*i,ctypes.POINTER(ctypes.c_short))[:1].pop() + tmpstr = ctypes.c_char_p(data+nbytes*i+2) + res.append( tmpstr.value.decode()[0:rbyte] ) + except ValueError: + res.append(None) + return res + +def _crow_nchar_to_python_block(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C nchar row to python row + """ + assert(nbytes is not None) + res=[] + if num_of_rows >= 0: + for i in range(abs(num_of_rows)): + try: + tmpstr = ctypes.c_char_p(data+nbytes*i+2) + res.append( tmpstr.value.decode() ) + except ValueError: + res.append(None) + else: + for i in range(abs(num_of_rows)): + try: + res.append( (ctypes.cast(data+nbytes*i+2, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) + except ValueError: + res.append(None) + return res + +_CONVERT_FUNC = { + FieldType.C_BOOL: _crow_bool_to_python, + FieldType.C_TINYINT : _crow_tinyint_to_python, + FieldType.C_SMALLINT : _crow_smallint_to_python, + FieldType.C_INT : _crow_int_to_python, + FieldType.C_BIGINT : _crow_bigint_to_python, + FieldType.C_FLOAT : _crow_float_to_python, + FieldType.C_DOUBLE : _crow_double_to_python, + FieldType.C_BINARY: _crow_binary_to_python, + FieldType.C_TIMESTAMP : _crow_timestamp_to_python, + FieldType.C_NCHAR : _crow_nchar_to_python +} + +_CONVERT_FUNC_BLOCK = { + FieldType.C_BOOL: _crow_bool_to_python, + FieldType.C_TINYINT : _crow_tinyint_to_python, + FieldType.C_SMALLINT : _crow_smallint_to_python, + FieldType.C_INT : _crow_int_to_python, + FieldType.C_BIGINT : _crow_bigint_to_python, + FieldType.C_FLOAT : _crow_float_to_python, + FieldType.C_DOUBLE : _crow_double_to_python, + FieldType.C_BINARY: _crow_binary_to_python_block, + FieldType.C_TIMESTAMP : _crow_timestamp_to_python, + FieldType.C_NCHAR : _crow_nchar_to_python_block +} + +# Corresponding TAOS_FIELD structure in C +class TaosField(ctypes.Structure): + _fields_ = [('name', ctypes.c_char * 65), + ('type', ctypes.c_char), + ('bytes', ctypes.c_short)] + +# C interface class +class CTaosInterface(object): + + libtaos = ctypes.CDLL('libtaos.dylib') + + libtaos.taos_fetch_fields.restype = ctypes.POINTER(TaosField) + libtaos.taos_init.restype = None + libtaos.taos_connect.restype = ctypes.c_void_p + #libtaos.taos_use_result.restype = ctypes.c_void_p + libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) + libtaos.taos_errstr.restype = ctypes.c_char_p + libtaos.taos_subscribe.restype = ctypes.c_void_p + libtaos.taos_consume.restype = ctypes.c_void_p + libtaos.taos_fetch_lengths.restype = ctypes.c_void_p + libtaos.taos_free_result.restype = None + libtaos.taos_errno.restype = ctypes.c_int + libtaos.taos_query.restype = ctypes.POINTER(ctypes.c_void_p) + + def __init__(self, config=None): + ''' + Function to initialize the class + @host : str, hostname to connect + @user : str, username to connect to server + @password : str, password to connect to server + @db : str, default db to use when log in + @config : str, config directory + + @rtype : None + ''' + if config is None: + self._config = ctypes.c_char_p(None) + else: + try: + self._config = ctypes.c_char_p(config.encode('utf-8')) + except AttributeError: + raise AttributeError("config is expected as a str") + + if config != None: + CTaosInterface.libtaos.taos_options(3, self._config) + + CTaosInterface.libtaos.taos_init() + + @property + def config(self): + """ Get current config + """ + return self._config + + def connect(self, host=None, user="root", password="taosdata", db=None, port=0): + ''' + Function to connect to server + + @rtype: c_void_p, TDengine handle + ''' + # host + try: + _host = ctypes.c_char_p(host.encode( + "utf-8")) if host != None else ctypes.c_char_p(None) + except AttributeError: + raise AttributeError("host is expected as a str") + + # user + try: + _user = ctypes.c_char_p(user.encode("utf-8")) + except AttributeError: + raise AttributeError("user is expected as a str") + + # password + try: + _password = ctypes.c_char_p(password.encode("utf-8")) + except AttributeError: + raise AttributeError("password is expected as a str") + + # db + try: + _db = ctypes.c_char_p( + db.encode("utf-8")) if db != None else ctypes.c_char_p(None) + except AttributeError: + raise AttributeError("db is expected as a str") + + # port + try: + _port = ctypes.c_int(port) + except TypeError: + raise TypeError("port is expected as an int") + + connection = ctypes.c_void_p(CTaosInterface.libtaos.taos_connect( + _host, _user, _password, _db, _port)) + + if connection.value == None: + print('connect to TDengine failed') + raise ConnectionError("connect to TDengine failed") + # sys.exit(1) + #else: + # print('connect to TDengine success') + + return connection + + @staticmethod + def close(connection): + '''Close the TDengine handle + ''' + CTaosInterface.libtaos.taos_close(connection) + #print('connection is closed') + + @staticmethod + def query(connection, sql): + '''Run SQL + + @sql: str, sql string to run + + @rtype: 0 on success and -1 on failure + ''' + try: + return CTaosInterface.libtaos.taos_query(connection, ctypes.c_char_p(sql.encode('utf-8'))) + except AttributeError: + raise AttributeError("sql is expected as a string") + # finally: + # CTaosInterface.libtaos.close(connection) + + @staticmethod + def affectedRows(result): + """The affected rows after runing query + """ + return CTaosInterface.libtaos.taos_affected_rows(result) + + @staticmethod + def subscribe(connection, restart, topic, sql, interval): + """Create a subscription + @restart boolean, + @sql string, sql statement for data query, must be a 'select' statement. + @topic string, name of this subscription + """ + return ctypes.c_void_p(CTaosInterface.libtaos.taos_subscribe( + connection, + 1 if restart else 0, + ctypes.c_char_p(topic.encode('utf-8')), + ctypes.c_char_p(sql.encode('utf-8')), + None, + None, + interval)) + + @staticmethod + def consume(sub): + """Consume data of a subscription + """ + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_consume(sub)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.libtaos.taos_num_fields(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + return result, fields + + @staticmethod + def unsubscribe(sub, keepProgress): + """Cancel a subscription + """ + CTaosInterface.libtaos.taos_unsubscribe(sub, 1 if keepProgress else 0) + + @staticmethod + def useResult(result): + '''Use result after calling self.query + ''' + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.fieldsCount(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + + return fields + + @staticmethod + def fetchBlock(result, fields): + pblock = ctypes.c_void_p(0) + num_of_rows = CTaosInterface.libtaos.taos_fetch_block( + result, ctypes.byref(pblock)) + if num_of_rows == 0: + return None, 0 + isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) + blocks = [None] * len(fields) + fieldL = CTaosInterface.libtaos.taos_fetch_lengths(result) + fieldLen = [ele for ele in ctypes.cast(fieldL, ctypes.POINTER(ctypes.c_int))[:len(fields)]] + for i in range(len(fields)): + data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] + if fields[i]['type'] not in _CONVERT_FUNC_BLOCK: + raise DatabaseError("Invalid data type returned from database") + blocks[i] = _CONVERT_FUNC_BLOCK[fields[i]['type']](data, num_of_rows, fieldLen[i], isMicro) + + return blocks, abs(num_of_rows) + @staticmethod + def fetchRow(result, fields): + pblock = ctypes.c_void_p(0) + pblock = CTaosInterface.libtaos.taos_fetch_row(result) + if pblock : + num_of_rows = 1 + isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) + blocks = [None] * len(fields) + fieldL = CTaosInterface.libtaos.taos_fetch_lengths(result) + fieldLen = [ele for ele in ctypes.cast(fieldL, ctypes.POINTER(ctypes.c_int))[:len(fields)]] + for i in range(len(fields)): + data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] + if fields[i]['type'] not in _CONVERT_FUNC: + raise DatabaseError("Invalid data type returned from database") + if data is None: + blocks[i] = [None] + else: + blocks[i] = _CONVERT_FUNC[fields[i]['type']](data, num_of_rows, fieldLen[i], isMicro) + else: + return None, 0 + return blocks, abs(num_of_rows) + + @staticmethod + def freeResult(result): + CTaosInterface.libtaos.taos_free_result(result) + result.value = None + + @staticmethod + def fieldsCount(result): + return CTaosInterface.libtaos.taos_field_count(result) + + @staticmethod + def fetchFields(result): + return CTaosInterface.libtaos.taos_fetch_fields(result) + + # @staticmethod + # def fetchRow(result, fields): + # l = [] + # row = CTaosInterface.libtaos.taos_fetch_row(result) + # if not row: + # return None + + # for i in range(len(fields)): + # l.append(CTaosInterface.getDataValue( + # row[i], fields[i]['type'], fields[i]['bytes'])) + + # return tuple(l) + + # @staticmethod + # def getDataValue(data, dtype, byte): + # ''' + # ''' + # if not data: + # return None + + # if (dtype == CTaosInterface.TSDB_DATA_TYPE_BOOL): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TINYINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_SMALLINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY): + # return (ctypes.cast(data, ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00') + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR): + # return (ctypes.cast(data, ctypes.c_char_p).value).rstrip('\x00') + + @staticmethod + def errno(result): + """Return the error number. + """ + return CTaosInterface.libtaos.taos_errno(result) + + @staticmethod + def errStr(result): + """Return the error styring + """ + return CTaosInterface.libtaos.taos_errstr(result).decode('utf-8') + + +if __name__ == '__main__': + cinter = CTaosInterface() + conn = cinter.connect() + result = cinter.query(conn, 'show databases') + + print('Query Affected rows: {}'.format(cinter.affectedRows(result))) + + fields = CTaosInterface.useResult(result) + + data, num_of_rows = CTaosInterface.fetchBlock(result, fields) + + print(data) + + cinter.freeResult(result) + cinter.close(conn) diff --git a/src/connector/python/osx/python3/taos/connection.py b/src/connector/python/osx/python3/taos/connection.py new file mode 100644 index 0000000000..552250f116 --- /dev/null +++ b/src/connector/python/osx/python3/taos/connection.py @@ -0,0 +1,86 @@ +from .cursor import TDengineCursor +from .subscription import TDengineSubscription +from .cinterface import CTaosInterface + +class TDengineConnection(object): + """ TDengine connection object + """ + def __init__(self, *args, **kwargs): + self._conn = None + self._host = None + self._user = "root" + self._password = "taosdata" + self._database = None + self._port = 0 + self._config = None + self._chandle = None + + self.config(**kwargs) + + def config(self, **kwargs): + # host + if 'host' in kwargs: + self._host = kwargs['host'] + + # user + if 'user' in kwargs: + self._user = kwargs['user'] + + # password + if 'password' in kwargs: + self._password = kwargs['password'] + + # database + if 'database' in kwargs: + self._database = kwargs['database'] + + # port + if 'port' in kwargs: + self._port = kwargs['port'] + + # config + if 'config' in kwargs: + self._config = kwargs['config'] + + self._chandle = CTaosInterface(self._config) + self._conn = self._chandle.connect(self._host, self._user, self._password, self._database, self._port) + + def close(self): + """Close current connection. + """ + return CTaosInterface.close(self._conn) + + def subscribe(self, restart, topic, sql, interval): + """Create a subscription. + """ + if self._conn is None: + return None + sub = CTaosInterface.subscribe(self._conn, restart, topic, sql, interval) + return TDengineSubscription(sub) + + def cursor(self): + """Return a new Cursor object using the connection. + """ + return TDengineCursor(self) + + def commit(self): + """Commit any pending transaction to the database. + + Since TDengine do not support transactions, the implement is void functionality. + """ + pass + + def rollback(self): + """Void functionality + """ + pass + + def clear_result_set(self): + """Clear unused result set on this connection. + """ + pass + +if __name__ == "__main__": + conn = TDengineConnection(host='192.168.1.107') + conn.close() + print("Hello world") \ No newline at end of file diff --git a/src/connector/python/osx/python3/taos/constants.py b/src/connector/python/osx/python3/taos/constants.py new file mode 100644 index 0000000000..feb7050a40 --- /dev/null +++ b/src/connector/python/osx/python3/taos/constants.py @@ -0,0 +1,33 @@ +"""Constants in TDengine python +""" + +from .dbapi import * + +class FieldType(object): + """TDengine Field Types + """ + # type_code + C_NULL = 0 + C_BOOL = 1 + C_TINYINT = 2 + C_SMALLINT = 3 + C_INT = 4 + C_BIGINT = 5 + C_FLOAT = 6 + C_DOUBLE = 7 + C_BINARY = 8 + C_TIMESTAMP = 9 + C_NCHAR = 10 + # NULL value definition + # NOTE: These values should change according to C definition in tsdb.h + C_BOOL_NULL = 0x02 + C_TINYINT_NULL = -128 + C_SMALLINT_NULL = -32768 + C_INT_NULL = -2147483648 + C_BIGINT_NULL = -9223372036854775808 + C_FLOAT_NULL = float('nan') + C_DOUBLE_NULL = float('nan') + C_BINARY_NULL = bytearray([int('0xff', 16)]) + # Timestamp precision definition + C_TIMESTAMP_MILLI = 0 + C_TIMESTAMP_MICRO = 1 diff --git a/src/connector/python/osx/python3/taos/cursor.py b/src/connector/python/osx/python3/taos/cursor.py new file mode 100644 index 0000000000..f972d2ff07 --- /dev/null +++ b/src/connector/python/osx/python3/taos/cursor.py @@ -0,0 +1,271 @@ +from .cinterface import CTaosInterface +from .error import * +from .constants import FieldType +import threading + +# querySeqNum = 0 + +class TDengineCursor(object): + """Database cursor which is used to manage the context of a fetch operation. + + Attributes: + .description: Read-only attribute consists of 7-item sequences: + + > name (mondatory) + > type_code (mondatory) + > display_size + > internal_size + > precision + > scale + > null_ok + + This attribute will be None for operations that do not return rows or + if the cursor has not had an operation invoked via the .execute*() method yet. + + .rowcount:This read-only attribute specifies the number of rows that the last + .execute*() produced (for DQL statements like SELECT) or affected + """ + + def __init__(self, connection=None): + self._description = [] + self._rowcount = -1 + self._connection = None + self._result = None + self._fields = None + self._block = None + self._block_rows = -1 + self._block_iter = 0 + self._affected_rows = 0 + self._logfile = "" + self._threadId = threading.get_ident() + + if connection is not None: + self._connection = connection + + def __iter__(self): + return self + + def __next__(self): + if self._result is None or self._fields is None: + raise OperationalError("Invalid use of fetch iterator") + + if self._block_rows <= self._block_iter: + block, self._block_rows = CTaosInterface.fetchRow( + self._result, self._fields) + if self._block_rows == 0: + raise StopIteration + self._block = list(map(tuple, zip(*block))) + self._block_iter = 0 + + data = self._block[self._block_iter] + self._block_iter += 1 + + return data + + @property + def description(self): + """Return the description of the object. + """ + return self._description + + @property + def rowcount(self): + """Return the rowcount of the object + """ + return self._rowcount + + @property + def affected_rows(self): + """Return the rowcount of insertion + """ + return self._affected_rows + + def callproc(self, procname, *args): + """Call a stored database procedure with the given name. + + Void functionality since no stored procedures. + """ + pass + + def log(self, logfile): + self._logfile = logfile + + def close(self): + """Close the cursor. + """ + if self._connection is None: + return False + + self._reset_result() + self._connection = None + + return True + + def execute(self, operation, params=None): + """Prepare and execute a database operation (query or command). + """ + # if threading.get_ident() != self._threadId: + # info ="Cursor execute:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident()) + # raise OperationalError(info) + # print(info) + # return None + + if not operation: + return None + + if not self._connection: + # TODO : change the exception raised here + raise ProgrammingError("Cursor is not connected") + + self._reset_result() + + stmt = operation + if params is not None: + pass + + # global querySeqNum + # querySeqNum += 1 + # localSeqNum = querySeqNum # avoid raice condition + # print(" >> Exec Query ({}): {}".format(localSeqNum, str(stmt))) + self._result = CTaosInterface.query(self._connection._conn, stmt) + # print(" << Query ({}) Exec Done".format(localSeqNum)) + if (self._logfile): + with open(self._logfile, "a") as logfile: + logfile.write("%s;\n" % operation) + + errno = CTaosInterface.libtaos.taos_errno(self._result) + if errno == 0: + if CTaosInterface.fieldsCount(self._result) == 0: + self._affected_rows += CTaosInterface.affectedRows( + self._result ) + return CTaosInterface.affectedRows(self._result ) + else: + self._fields = CTaosInterface.useResult( + self._result) + return self._handle_result() + else: + raise ProgrammingError( + CTaosInterface.errStr( + self._result), errno) + + def executemany(self, operation, seq_of_parameters): + """Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings found in the sequence seq_of_parameters. + """ + pass + + def fetchone(self): + """Fetch the next row of a query result set, returning a single sequence, or None when no more data is available. + """ + pass + + def fetchmany(self): + pass + + def istype(self, col, dataType): + if (dataType.upper() == "BOOL"): + if (self._description[col][1] == FieldType.C_BOOL): + return True + if (dataType.upper() == "TINYINT"): + if (self._description[col][1] == FieldType.C_TINYINT): + return True + if (dataType.upper() == "INT"): + if (self._description[col][1] == FieldType.C_INT): + return True + if (dataType.upper() == "BIGINT"): + if (self._description[col][1] == FieldType.C_INT): + return True + if (dataType.upper() == "FLOAT"): + if (self._description[col][1] == FieldType.C_FLOAT): + return True + if (dataType.upper() == "DOUBLE"): + if (self._description[col][1] == FieldType.C_DOUBLE): + return True + if (dataType.upper() == "BINARY"): + if (self._description[col][1] == FieldType.C_BINARY): + return True + if (dataType.upper() == "TIMESTAMP"): + if (self._description[col][1] == FieldType.C_TIMESTAMP): + return True + if (dataType.upper() == "NCHAR"): + if (self._description[col][1] == FieldType.C_NCHAR): + return True + + return False + + def fetchall_row(self): + """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences (e.g. a list of tuples). Note that the cursor's arraysize attribute can affect the performance of this operation. + """ + if self._result is None or self._fields is None: + raise OperationalError("Invalid use of fetchall") + + buffer = [[] for i in range(len(self._fields))] + self._rowcount = 0 + while True: + block, num_of_fields = CTaosInterface.fetchRow(self._result, self._fields) + errno = CTaosInterface.libtaos.taos_errno(self._result) + if errno != 0: + raise ProgrammingError(CTaosInterface.errStr(self._result), errno) + if num_of_fields == 0: + break + self._rowcount += num_of_fields + for i in range(len(self._fields)): + buffer[i].extend(block[i]) + return list(map(tuple, zip(*buffer))) + + def fetchall(self): + if self._result is None or self._fields is None: + raise OperationalError("Invalid use of fetchall") + + buffer = [[] for i in range(len(self._fields))] + self._rowcount = 0 + while True: + block, num_of_fields = CTaosInterface.fetchBlock(self._result, self._fields) + errno = CTaosInterface.libtaos.taos_errno(self._result) + if errno != 0: + raise ProgrammingError(CTaosInterface.errStr(self._result), errno) + if num_of_fields == 0: break + self._rowcount += num_of_fields + for i in range(len(self._fields)): + buffer[i].extend(block[i]) + return list(map(tuple, zip(*buffer))) + def nextset(self): + """ + """ + pass + + def setinputsize(self, sizes): + pass + + def setutputsize(self, size, column=None): + pass + + def _reset_result(self): + """Reset the result to unused version. + """ + self._description = [] + self._rowcount = -1 + if self._result is not None: + CTaosInterface.freeResult(self._result) + self._result = None + self._fields = None + self._block = None + self._block_rows = -1 + self._block_iter = 0 + self._affected_rows = 0 + + def _handle_result(self): + """Handle the return result from query. + """ + # if threading.get_ident() != self._threadId: + # info = "Cursor handleresult:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident()) + # raise OperationalError(info) + # print(info) + # return None + + self._description = [] + for ele in self._fields: + self._description.append( + (ele['name'], ele['type'], None, None, None, None, False)) + + return self._result + diff --git a/src/connector/python/osx/python3/taos/dbapi.py b/src/connector/python/osx/python3/taos/dbapi.py new file mode 100644 index 0000000000..f1c22bdb51 --- /dev/null +++ b/src/connector/python/osx/python3/taos/dbapi.py @@ -0,0 +1,38 @@ +"""Type Objects and Constructors. +""" + +import time +import datetime + +class DBAPITypeObject(object): + def __init__(self, *values): + self.values = values + + def __com__(self, other): + if other in self.values: + return 0 + if other < self.values: + return 1 + else: + return -1 + +Date = datetime.date +Time = datetime.time +Timestamp = datetime.datetime + +def DataFromTicks(ticks): + return Date(*time.localtime(ticks)[:3]) + +def TimeFromTicks(ticks): + return Time(*time.localtime(ticks)[3:6]) + +def TimestampFromTicks(ticks): + return Timestamp(*time.localtime(ticks)[:6]) + +Binary = bytes + +# STRING = DBAPITypeObject(*constants.FieldType.get_string_types()) +# BINARY = DBAPITypeObject(*constants.FieldType.get_binary_types()) +# NUMBER = BAPITypeObject(*constants.FieldType.get_number_types()) +# DATETIME = DBAPITypeObject(*constants.FieldType.get_timestamp_types()) +# ROWID = DBAPITypeObject() \ No newline at end of file diff --git a/src/connector/python/osx/python3/taos/error.py b/src/connector/python/osx/python3/taos/error.py new file mode 100644 index 0000000000..24508a72ed --- /dev/null +++ b/src/connector/python/osx/python3/taos/error.py @@ -0,0 +1,57 @@ +"""Python exceptions +""" + +class Error(Exception): + def __init__(self, msg=None, errno=None): + self.msg = msg + self._full_msg = self.msg + self.errno = errno + + def __str__(self): + return self._full_msg + +class Warning(Exception): + """Exception raised for important warnings like data truncations while inserting. + """ + pass + +class InterfaceError(Error): + """Exception raised for errors that are related to the database interface rather than the database itself. + """ + pass + +class DatabaseError(Error): + """Exception raised for errors that are related to the database. + """ + pass + +class DataError(DatabaseError): + """Exception raised for errors that are due to problems with the processed data like division by zero, numeric value out of range. + """ + pass + +class OperationalError(DatabaseError): + """Exception raised for errors that are related to the database's operation and not necessarily under the control of the programmer + """ + pass + + +class IntegrityError(DatabaseError): + """Exception raised when the relational integrity of the database is affected. + """ + pass + +class InternalError(DatabaseError): + """Exception raised when the database encounters an internal error. + """ + pass + +class ProgrammingError(DatabaseError): + """Exception raised for programming errors. + """ + pass + +class NotSupportedError(DatabaseError): + """Exception raised in case a method or database API was used which is not supported by the database,. + """ + pass \ No newline at end of file diff --git a/src/connector/python/osx/python3/taos/subscription.py b/src/connector/python/osx/python3/taos/subscription.py new file mode 100644 index 0000000000..d3cf10d5ad --- /dev/null +++ b/src/connector/python/osx/python3/taos/subscription.py @@ -0,0 +1,52 @@ +from .cinterface import CTaosInterface +from .error import * + +class TDengineSubscription(object): + """TDengine subscription object + """ + def __init__(self, sub): + self._sub = sub + + + def consume(self): + """Consume rows of a subscription + """ + if self._sub is None: + raise OperationalError("Invalid use of consume") + + result, fields = CTaosInterface.consume(self._sub) + buffer = [[] for i in range(len(fields))] + while True: + block, num_of_fields = CTaosInterface.fetchBlock(result, fields) + if num_of_fields == 0: break + for i in range(len(fields)): + buffer[i].extend(block[i]) + + self.fields = fields + return list(map(tuple, zip(*buffer))) + + + def close(self, keepProgress = True): + """Close the Subscription. + """ + if self._sub is None: + return False + + CTaosInterface.unsubscribe(self._sub, keepProgress) + return True + + +if __name__ == '__main__': + from .connection import TDengineConnection + conn = TDengineConnection(host="127.0.0.1", user="root", password="taosdata", database="test") + + # Generate a cursor object to run SQL commands + sub = conn.subscribe(True, "test", "select * from meters;", 1000) + + for i in range(0,10): + data = sub.consume() + for d in data: + print(d) + + sub.close() + conn.close() \ No newline at end of file -- GitLab