TDengine Python Client Library

taospy is the official Python client library for TDengine. taospy provides a rich API that makes it easy for Python applications to use TDengine.

The source code for the Python client library is hosted on GitHub.

Connection types

taospy mainly provides 3 connection types, among which we recommend using websocket connection.

• Native connection, which correspond to the taos modules of the taospy package, connects to TDengine instances natively through the TDengine client driver (taosc), supporting data writing, querying, subscriptions, schemaless writing, and bind interface.
• REST connection, which correspond to the taosrest modules of the taospy package, which is implemented through taosAdapter. Some features like schemaless and subscriptions are not supported.
• Websocket connection taos-ws-py is an optional package to enable using WebSocket to connect TDengine, which is implemented through taosAdapter. The set of features implemented by the WebSocket connection differs slightly from those implemented by the native connection.

For a detailed introduction of the connection types, please refer to: Establish Connection

In addition to wrapping the native and REST interfaces, taospy also provides a set of programming interfaces that conforms to the Python Data Access Specification (PEP 249). It is easy to integrate taospy with many third-party tools, such as SQLAlchemy and pandas.

The direct connection to the server using the native interface provided by the client driver is referred to hereinafter as a "native connection"; the connection to the server using the REST or WebSocket interface provided by taosAdapter is referred to hereinafter as a "REST connection" or "WebSocket connection".

Supported platforms

• The supported platforms for the native connection are the same as the ones supported by the TDengine client.
• REST connections are supported on all platforms that can run Python.

Supported features

• Native connections support all the core features of TDengine, including connection management, SQL execution, bind interface, subscriptions, and schemaless writing.
• REST connections support features such as connection management and SQL execution. (SQL execution allows you to: manage databases, tables, and supertables, write data, query data, create continuous queries, etc.).

Version selection

We recommend using the latest version of taospy, regardless of the version of TDengine.

Python Client Library Version	major changes
2.7.16	add subscription configuration (session.timeout.ms, max.poll.interval.ms)
2.7.15	added support for VARBINARY and GEOMETRY types
2.7.14	fix known issues
2.7.13	add TMQ synchronous submission offset interface
2.7.12	1. added support for varbinary type (STMT does not yet support) 2. improved query performance (thanks to contributor hadrianl)
2.7.9	support for getting assignment and seek function on subscription
2.7.8	add execute_many method

Python Websocket Connection Version	major changes
0.3.5	1. added support for VARBINARY and GEOMETRY types 2. Fix known issues
0.3.2	1. optimize WebSocket SQL query and insertion performance 2. Fix known issues 3. Modify the readme and document
0.2.9	bugs fixes
0.2.5	1. support for getting assignment and seek function on subscription 2. support schemaless 3. support STMT
0.2.4	support unsubscribe on subscription

Handling Exceptions

There are 4 types of exception in python client library.

• The exception of Python client library itself.
• The exception of native library.
• The exception of websocket
• The exception of subscription.
• The exception of other TDengine function modules.

Error Type	Description	Suggested Actions
InterfaceError	the native library is too old that it cannot support the function	please check the TDengine client version
ConnectionError	connection error	please check TDengine's status and the connection params
DatabaseError	database error	please upgrade Python client library to latest
OperationalError	operation error
ProgrammingError
StatementError	the exception of stmt
ResultError
SchemalessError	the exception of stmt schemaless
TmqError	the exception of stmt tmq

It usually uses try-expect to handle exceptions in python. For exception handling, please refer to Python Errors and Exceptions Documentation.

All exceptions from the Python client library are thrown directly. Applications should handle these exceptions. For example:

import taos

try:
    conn = taos.connect()
    conn.execute("CREATE TABLE 123")  # wrong sql
except taos.Error as e:
    print(e)
    print("exception class: ", e.__class__.__name__)
    print("error number:", e.errno)
    print("error message:", e.msg)
except BaseException as other:
    print("exception occur")
    print(other)

# output:
# [0x0216]: syntax error near 'Incomplete SQL statement'
# exception class:  ProgrammingError
# error number: -2147483114
# error message: syntax error near 'Incomplete SQL statement'

view source code

TDengine DataType vs. Python DataType

TDengine currently supports timestamp, number, character, Boolean type, and the corresponding type conversion with Python is as follows:

TDengine DataType	Python DataType
TIMESTAMP	datetime
INT	int
BIGINT	int
FLOAT	float
DOUBLE	int
SMALLINT	int
TINYINT	int
BOOL	bool
BINARY	str
NCHAR	str
JSON	str

Installation Steps

Pre-installation preparation

1. Install Python. The recent taospy package requires Python 3.6.2+. The earlier versions of taospy require Python 3.7+. The taos-ws-py package requires Python 3.7+. If Python is not available on your system, refer to the Python BeginnersGuide to install it.
2. Install pip. In most cases, the Python installer comes with the pip utility. If not, please refer to pip documentation to install it. If you use a native connection, you will also need to Install Client Driver. The client install package includes the TDengine client dynamic link library (libtaos.so or taos.dll) and the TDengine CLI.

Install via pip

Uninstalling an older version

If you have installed an older version of the Python client library, please uninstall it beforehand.

pip3 uninstall taos taospy

note

Earlier TDengine client software includes the Python client library. If the Python client library is installed from the client package's installation directory, the corresponding Python package name is taos. So the above uninstall command includes taos, and it doesn't matter if it doesn't exist.

To install taospy

Install from PyPI

Install the latest version of:

pip3 install taospy

You can also specify a specific version to install:

pip3 install taospy==2.3.0

Install from GitHub

pip3 install git+https://github.com/taosdata/taos-connector-python.git

Install taos-ws-py (Optional)

The taos-ws-py package provides the way to access TDengine via WebSocket.

Install taos-ws-py with taospy

pip3 install taospy[ws]

Install taos-ws-py only

pip3 install taos-ws-py

Verify

native connection

For native connection, you need to verify that both the client driver and the Python client library itself are installed correctly. The client driver and Python client library have been installed properly if you can successfully import the taos module. In the Python Interactive Shell, you can type.

import taos

REST connection

For REST connections, verifying that the taosrest module can be imported successfully can be done in the Python Interactive Shell by typing.

import taosrest

WebSocket connection

For WebSocket connection, verifying that the taosws module can be imported successfully can be done in the Python Interactive Shell by typing.

import taosws

tip

If you have multiple versions of Python on your system, you may have various pip commands. Be sure to use the correct path for the pip command. Above, we installed the pip3 command, which rules out the possibility of using the pip corresponding to Python 2.x versions. However, if you have more than one version of Python 3.x on your system, you still need to check that the installation path is correct. The easiest way to verify this is to type pip3 install taospy again in the command, and it will print out the exact location of taospy, for example, on Windows.

C:\> pip3 install taospy
Looking in indexes: https://pypi.tuna.tsinghua.edu.cn/simple
Requirement already satisfied: taospy in c:\users\username\appdata\local\programs\python\python310\lib\site-packages (2.3.0)

Establishing a connection

Connectivity testing

Before establishing a connection with the client library, we recommend testing the connectivity of the local TDengine CLI to the TDengine cluster.

native connection

Ensure that the TDengine instance is up and that the FQDN of the machines in the cluster (the FQDN defaults to hostname if you are starting a stand-alone version) can be resolved locally, by testing with the ping command.

ping <FQDN>

Then test if the cluster can be appropriately connected with TDengine CLI:

taos -h <FQDN> -p <PORT>

The FQDN above can be the FQDN of any dnode in the cluster, and the PORT is the serverPort corresponding to this dnode.

REST connection

For REST connections, make sure the cluster and taosAdapter component, are running. This can be tested using the following curl command.

curl -u root:taosdata http://<FQDN>:<PORT>/rest/sql -d "select server_version()"

The FQDN above is the FQDN of the machine running taosAdapter, PORT is the port taosAdapter listening, default is 6041. If the test is successful, it will output the server version information, e.g.

{
  "code": 0,
  "column_meta": [
    [
      "server_version()",
      "VARCHAR",
      7
    ]
  ],
  "data": [
    [
      "3.0.0.0"
    ]
  ],
  "rows": 1
}

WebSocket connection

For WebSocket connection, make sure the cluster and taosAdapter component, are running. This can be testetd using the following curl command.

curl -i -N -d "show databases" -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -H "Connection: Upgrade" -H "Upgrade: websocket" -H "Host: <FQDN>:<PORT>" -H "Origin: http://<FQDN>:<PORT>" http://<FQDN>:<PORT>/rest/sql

The FQDN above is the FQDN of the machine running taosAdapter, PORT is the port taosAdapter listening, default is 6041.

If the test is successful, it will output the server version information, e.g.

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Tue, 21 Mar 2023 09:29:17 GMT
Transfer-Encoding: chunked

{"status":"succ","head":["server_version()"],"column_meta":[["server_version()",8,8]],"data":[["2.6.0.27"]],"rows":1}

Specify the Host and Properties to get the connection

The following example code assumes that TDengine is installed locally and that the default configuration is used for both FQDN and serverPort.

native connection

import taos

conn = taos.connect(
    host="localhost",
    user="root",
    password="taosdata",
    database="test",
    port=6030,
    config="/etc/taos",  # for windows the default value is C:\TDengine\cfg
    timezone="Asia/Shanghai",
)  # default your host's timezone

server_version = conn.server_info
print("server_version", server_version)
client_version = conn.client_info
print("client_version", client_version)  # 3.0.0.0

conn.close()

# possible output:
# 3.0.0.0
# 3.0.0.0

view source code

All arguments of the connect() function are optional keyword arguments. The following are the connection parameters specified.

• host : The FQDN of the node to connect to. There is no default value. If this parameter is not provided, the firstEP in the client configuration file will be connected.
• user : The TDengine user name. The default value is root.
• password : TDengine user password. The default value is taosdata.
• port : The starting port of the data node to connect to, i.e., the serverPort configuration. The default value is 6030, which will only take effect if the host parameter is provided.
• config : The path to the client configuration file. On Windows systems, the default is C:\TDengine\cfg. The default is /etc/taos/ on Linux/macOS.
• timezone : The timezone used to convert the TIMESTAMP data in the query results to python datetime objects. The default is the local timezone.

warning

config and timezone are both process-level configurations. We recommend that all connections made by a process use the same parameter values. Otherwise, unpredictable errors may occur.

tip

The connect() function returns a taos.TaosConnection instance. In client-side multi-threaded scenarios, we recommend that each thread request a separate connection instance rather than sharing a connection between multiple threads.

REST connection

from taosrest import connect, TaosRestConnection, TaosRestCursor

conn = connect(url="http://localhost:6041",
               user="root",
               password="taosdata",
               timeout=30)

view source code

All arguments to the connect() function are optional keyword arguments. The following are the connection parameters specified.

• url: The URL of taosAdapter REST service. The default is http://localhost:6041.
• user: TDengine user name. The default is root.
• password: TDengine user password. The default is taosdata.
• timeout: HTTP request timeout. Enter a value in seconds. The default is socket._GLOBAL_DEFAULT_TIMEOUT. Usually, no configuration is needed.

WebSocket connection

import taosws

def create_connection():
    conn = None
    host = "localhost"
    port = 6041
    try:
        conn = taosws.connect(
            user="root",
            password="taosdata",
            host=host,
            port=port,
        )
        print(f"Connected to {host}:{port} successfully.");
    except Exception as err:
        print(f"Failed to connect to {host}:{port} , ErrMessage:{err}")
        raise err
    return conn

view source code

The parameter of connect() is the url of TDengine, and the protocol is taosws or ws.

Priority of configuration parameters

If the configuration parameters are duplicated in the parameters or client configuration file, the priority of the parameters, from highest to lowest, are as follows:

1. Parameters in connect function.
2. the configuration file taos.cfg of the TDengine client driver when using a native connection.

Usage examples

Create database and tables

native connection

import taos

conn = None
host = "localhost"
port = 6030
try:
    conn = taos.connect(host=host,
                        port=port,
                        user="root",
                        password="taosdata")

    # create database
    rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS power")
    print(f"Create database power successfully, rowsAffected: {rowsAffected}");
    
    # create super table
    rowsAffected = conn.execute(
        "CREATE TABLE IF NOT EXISTS power.meters (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
    )
    print(f"Create stable power.meters successfully, rowsAffected: {rowsAffected}");

except Exception as err:
    print(f"Failed to create database power or stable meters, ErrMessage:{err}") 
    raise err
finally:
    if conn:
        conn.close()       

view source code

REST connection

import taosrest

conn = None
url = "http://localhost:6041"
try:
    conn = taosrest.connect(url=url,
                   user="root",
                   password="taosdata",
                   timeout=30)

    # create database
    rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS power")
    print(f"Create database power successfully, rowsAffected: {rowsAffected}");

    # create super table
    rowsAffected = conn.execute(
        f"CREATE TABLE IF NOT EXISTS power.meters (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
    )
    print(f"Create stable power.meters successfully, rowsAffected: {rowsAffected}");

except Exception as err:
    print(f"Failed to create database power or stable meters, ErrMessage:{err}") 
    raise err
finally:
    if conn:
        conn.close()

view source code

WebSocket connection

import taosws

conn = None
host = "localhost"
port = 6041
try:
    conn = taosws.connect(user="root",
                          password="taosdata",
                          host=host,
                          port=port)

    # create database
    rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS power")
    print(f"Create database power successfully, rowsAffected: {rowsAffected}");
    
    # create super table
    rowsAffected = conn.execute(
        "CREATE TABLE IF NOT EXISTS power.meters (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
    )
    print(f"Create stable power.meters successfully, rowsAffected: {rowsAffected}");

except Exception as err:
    print(f"Failed to create database power or stable meters, ErrMessage:{err}") 
    raise err
finally:
    if conn:
        conn.close()

view source code

Insert data

native connection

import taos

conn = None
host = "localhost"
port = 6030
try:
    conn = taos.connect(host=host,
                        port=port,
                        user="root",
                        password="taosdata")

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location)  TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

REST connection

import taosrest

conn = None
url="http://localhost:6041"
try:
    conn = taosrest.connect(url=url,
                   user="root",
                   password="taosdata",
                   timeout=30)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql:{sql}, ErrMessage:{err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

WebSocket connection

import taosws

conn = None
host="localhost"
port=6041
try:
    conn = taosws.connect(user="root",
                          password="taosdata",
                          host=host,
                          port=port)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

NOW is an internal function. The default is the current time of the client's computer. NOW + 1s represents the current time of the client plus 1 second, followed by the number representing the unit of time: a (milliseconds), s (seconds), m (minutes), h (hours), d (days), w (weeks), n (months), y (years).

Querying Data

native connection

The query method of the TaosConnection class can be used to query data and return the result data of type TaosResult.

import taos

conn = None
host = "localhost"
port = 6030
try:
    conn = taos.connect(host=host,
                        port=port,
                        user="root",
                        password="taosdata")

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location)  TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

tip

The queried results can only be fetched once. For example, only one of fetch_all() and fetch_all_into_dict() can be used in the example above. Repeated fetches will result in an empty list.

REST connection

The RestClient class is a direct wrapper for the REST API. It contains only a sql() method for executing arbitrary SQL statements and returning the result.

import taosrest

conn = None
url="http://localhost:6041"
try:
    conn = taosrest.connect(url=url,
                   user="root",
                   password="taosdata",
                   timeout=30)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql:{sql}, ErrMessage:{err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

For a more detailed description of the sql() method, please refer to RestClient.

WebSocket connection

The query method of the TaosConnection class can be used to query data and return the result data of type TaosResult.

import taosws

conn = None
host="localhost"
port=6041
try:
    conn = taosws.connect(user="root",
                          password="taosdata",
                          host=host,
                          port=port)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

Execute SQL with reqId

The reqId is very similar to TraceID in distributed tracing systems. In a distributed system, a request may need to pass through multiple services or modules to be completed. The reqId is used to identify and associate all related operations of this request, allowing us to track and understand the complete execution path of the request. Here are some primary usage of reqId:

• Request Tracing: By associating the same reqId with all related operations of a request, we can trace the complete path of the request within the system.
• Performance Analysis: By analyzing a request's reqId, we can understand the processing time of the request across various services or modules, thereby identifying performance bottlenecks.
• Fault Diagnosis: When a request fails, we can identify the location of the issue by examining the reqId associated with that request.

If the user does not set a reqId, the client library will generate one randomly internally, but it is still recommended for the user to set it, as it can better associate with the user's request.

native connection

As the way to connect introduced above but add req_id argument.

import taos

conn = None
host = "localhost"
port = 6030
try:
    conn = taos.connect(host=host,
                        port=port,
                        user="root",
                        password="taosdata")

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location)  TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

REST connection

As the way to connect introduced above but add req_id argument.

import taosrest

conn = None
url="http://localhost:6041"
try:
    conn = taosrest.connect(url=url,
                   user="root",
                   password="taosdata",
                   timeout=30)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql:{sql}, ErrMessage:{err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

WebSocket connection

As the way to connect introduced above but add req_id argument.

import taosws

conn = None
host="localhost"
port=6041
try:
    conn = taosws.connect(user="root",
                          password="taosdata",
                          host=host,
                          port=port)

    sql = """
        INSERT INTO 
        power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
            VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
            (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
        power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
            VALUES (NOW + 1a, 10.30000, 218, 0.25000)
        """
    affectedRows = conn.execute(sql)
    print(f"Successfully inserted {affectedRows} rows to power.meters.")

except Exception as err:
    print(f"Failed to insert data to power.meters, sql: {sql}, ErrMessage: {err}.")
    raise err
finally:
    if conn:
        conn.close()

view source code

Writing data via parameter binding

The Python client library provides a parameter binding api for inserting data. Similar to most databases, TDengine currently only supports the question mark ? to indicate the parameters to be bound.

native connection

sql = "INSERT INTO ? USING meters (groupid, location) TAGS(?,?) VALUES (?,?,?,?)"
stmt = conn.statement(sql)
for i in range(numOfSubTable):
    tbname = f"d_bind_{i}"

    tags = taos.new_bind_params(2)
    tags[0].int([i])
    tags[1].binary([f"location_{i}"])
    stmt.set_tbname_tags(tbname, tags)

    current = int(datetime.now().timestamp() * 1000)
    timestamps = []
    currents = []
    voltages = []
    phases = []

    for j in range (numOfRow):
        timestamps.append(current + i)
        currents.append(random.random() * 30)
        voltages.append(random.randint(100, 300))
        phases.append(random.random())

    params = taos.new_bind_params(4)
    params[0].timestamp(timestamps)
    params[1].float(currents)
    params[2].int(voltages)
    params[3].float(phases)
    stmt.bind_param_batch(params)
    stmt.execute()
    print(f"Successfully inserted to power.meters.")

except Exception as err:
print(f"Failed to insert to table meters using stmt, ErrMessage:{err}") 
raise err
finally:
if stmt:
    stmt.close()
if conn:    
    conn.close()

view source code

WebSocket connection

sql = "INSERT INTO ? USING meters (groupid, location) TAGS(?,?) VALUES (?,?,?,?)"
stmt = conn.statement()
stmt.prepare(sql)

for i in range(numOfSubTable):
    tbname = f"d_bind_{i}"

    tags = [
        taosws.int_to_tag(i),
        taosws.varchar_to_tag(f"location_{i}"),
    ]
    stmt.set_tbname_tags(tbname, tags)
    current = int(datetime.now().timestamp() * 1000)
    timestamps = []
    currents = []
    voltages = []
    phases = []
    for j in range (numOfRow):
        timestamps.append(current + i)
        currents.append(random.random() * 30)
        voltages.append(random.randint(100, 300))
        phases.append(random.random())

    stmt.bind_param(
        [
            taosws.millis_timestamps_to_column(timestamps),
            taosws.floats_to_column(currents),
            taosws.ints_to_column(voltages),
            taosws.floats_to_column(phases),
        ]
    )

    stmt.add_batch()
    stmt.execute()
        
    print(f"Successfully inserted to power.meters.")
        
except Exception as err:
print(f"Failed to insert to table meters using stmt, ErrMessage:{err}") 
raise err
finally:
if stmt:
    stmt.close()
if conn:    
    conn.close()

view source code

Schemaless Writing

Client library support schemaless insert.

native connection

import taos

lineDemo = [
    "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639"
]

telnetDemo = ["metric_telnet 1707095283260 4 host=host0 interface=eth0"]

jsonDemo = [
    '{"metric": "metric_json","timestamp": 1626846400,"value": 10.3, "tags": {"groupid": 2, "location": "California.SanFrancisco", "id": "d1001"}}'
]
host = "localhost"
port = 6030
try:
    conn = taos.connect(
        user="root",
        password="taosdata",
        host=host,
        port=port
    )

    conn.execute("CREATE DATABASE IF NOT EXISTS power")
    # change database. same as execute "USE db"
    conn.select_db("power")

    conn.schemaless_insert(
        lineDemo, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.MILLI_SECONDS
    )
    conn.schemaless_insert(
        telnetDemo, taos.SmlProtocol.TELNET_PROTOCOL, taos.SmlPrecision.MICRO_SECONDS
    )
    conn.schemaless_insert(
        jsonDemo, taos.SmlProtocol.JSON_PROTOCOL, taos.SmlPrecision.MILLI_SECONDS
    )
    print("Inserted data with schemaless successfully.");
except Exception as err:
    print(f"Failed to insert data with schemaless, ErrMessage:{err}")
    raise err
finally:
    if conn:
        conn.close()

view source code

WebSocket connection

import taosws

host = "localhost"
port = 6041
def prepare():
    conn = None
    try:
        conn = taosws.connect(user="root",
                            password="taosdata",
                            host=host,
                            port=port)

        # create database
        rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS power")
        assert rowsAffected == 0

    except Exception as err:
        print(f"Failed to create db and table, db addrr:{host}:{port} ; ErrMessage:{err}")
        raise err
    finally:
        if conn:
            conn.close()

def schemaless_insert():

    conn = None

    lineDemo = [
        "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639"
    ]

    telnetDemo = ["metric_telnet 1707095283260 4 host=host0 interface=eth0"]

    jsonDemo = [
        '{"metric": "metric_json","timestamp": 1626846400,"value": 10.3, "tags": {"groupid": 2, "location": "California.SanFrancisco", "id": "d1001"}}'
    ]

    try:
        conn = taosws.connect(user="root",
                              password="taosdata",
                              host=host,
                              port=port,
                              database='power')

        conn.schemaless_insert(
            lines = lineDemo,
            protocol = taosws.PySchemalessProtocol.Line,
            precision = taosws.PySchemalessPrecision.Millisecond,
            ttl=1,
            req_id=1,
        )

        conn.schemaless_insert(
            lines=telnetDemo,
            protocol=taosws.PySchemalessProtocol.Telnet,
            precision=taosws.PySchemalessPrecision.Microsecond,
            ttl=1,
            req_id=2,
        )

        conn.schemaless_insert(
            lines=jsonDemo,
            protocol=taosws.PySchemalessProtocol.Json,
            precision=taosws.PySchemalessPrecision.Millisecond,
            ttl=1,
            req_id=3,
        )
        print("Inserted data with schemaless successfully.");
    except Exception as err:
        print(f"Failed to insert data with schemaless, ErrMessage:{err}")
        raise err
    
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    prepare()
    schemaless_insert()
  

view source code

Schemaless with reqId

There is a optional parameter called req_id in schemaless_insert and schemaless_insert_raw method. This reqId can be used to request link tracing.

conn.schemaless_insert(
    lines=lineDemo,
    protocol=taos.SmlProtocol.LINE_PROTOCOL,
    precision=taos.SmlPrecision.NANO_SECONDS,
    req_id=1,
)

Data Subscription

Client library support data subscription. For more information about subscroption, please refer to Data Subscription.

Create a Topic

# create topic
conn.execute(
    f"CREATE TOPIC IF NOT EXISTS {topic} AS SELECT ts, current, voltage, phase, groupid, location FROM meters"
)

view source code

Create a Consumer

from taos.tmq import Consumer

def create_consumer():
    try:
        consumer = Consumer(
            {
                "group.id": groupId,
                "client.id": clientId,
                "td.connect.user": user,
                "td.connect.pass": password,
                "enable.auto.commit": autoCommitState,
                "auto.commit.interval.ms": autoCommitIntv,
                "auto.offset.reset": autoOffsetReset,
                "td.connect.ip": host,
                "td.connect.port": str(port),
            }
        )
        print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}")
        return consumer
    except Exception as err:
        print(f"Failed to create native consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err

view source code

Subscribe to a Topic

def subscribe(consumer):
    try:
        # subscribe to the topics
        consumer.subscribe(["topic_meters"])
        print("Subscribe topics successfully")
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        data = block.fetchall()
                        print(f"data: {data}")

    except Exception as err:
        print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err

view source code

Consume messages

#!/usr/bin/python3
import taos

db              = "power"
topic           = "topic_meters"
user            = "root"
password        = "taosdata"
host            = "localhost"
port            = 6030
groupId         = "group1"
clientId        = "1"  
tdConnWsScheme  = "ws"
autoOffsetReset = "latest"
autoCommitState = "true"
autoCommitIntv  = "1000"


def prepareMeta():
    conn = None
    try:
        conn = taos.connect(host=host, user=user, password=password, port=port)
        conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")

        # change database. same as execute "USE db"
        conn.select_db(db)

        # create super table
        conn.execute(
            "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)"
        )

        # create topic
        conn.execute(
            f"CREATE TOPIC IF NOT EXISTS {topic} AS SELECT ts, current, voltage, phase, groupid, location FROM meters"
        )
        sql = """
            INSERT INTO 
            power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
                VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
                (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
            power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
                VALUES (NOW + 1a, 10.30000, 218, 0.25000)
            """
        affectedRows = conn.execute(sql)
        print(f"Inserted into {affectedRows} rows to power.meters successfully.")
    except Exception as err:
        print(f"Failed to prepareMeta, host: {host}:{port}, db: {db}, topic: {topic}, ErrMessage:{err}.")
        raise err
    finally:
        if conn:
            conn.close()

from taos.tmq import Consumer

def create_consumer():
    try:
        consumer = Consumer(
            {
                "group.id": groupId,
                "client.id": clientId,
                "td.connect.user": user,
                "td.connect.pass": password,
                "enable.auto.commit": autoCommitState,
                "auto.commit.interval.ms": autoCommitIntv,
                "auto.offset.reset": autoOffsetReset,
                "td.connect.ip": host,
                "td.connect.port": str(port),
            }
        )
        print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}")
        return consumer
    except Exception as err:
        print(f"Failed to create native consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err


def subscribe(consumer):
    try:
        # subscribe to the topics
        consumer.subscribe(["topic_meters"])
        print("Subscribe topics successfully")
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        data = block.fetchall()
                        print(f"data: {data}")

    except Exception as err:
        print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err



def commit_offset(consumer):
    try:
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        print(block.fetchall())

                # after processing the data, commit the offset manually
                consumer.commit(records)
                print("Commit offset manually successfully.");

    except Exception as err:
        print(f"Failed to commit offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err


def seek_offset(consumer):
    try:
        assignments = consumer.assignment()
        if assignments:
            for partition in assignments:
                partition.offset = 0
                consumer.seek(partition)
                print(f"Assignment seek to beginning successfully.")
    except Exception as err:
        print(f"Failed to seek offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err

def unsubscribe(consumer):
    try:
        consumer.unsubscribe()
        print("Consumer unsubscribed successfully.");
    except Exception as err:
        print(f"Failed to unsubscribe consumer. topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    finally:
        if consumer:
            consumer.close()
            print("Consumer closed successfully.");        

if __name__ == "__main__":
    consumer = None
    try:
        prepareMeta()
        consumer = create_consumer()
        subscribe(consumer)
        seek_offset(consumer)
        commit_offset(consumer)
    finally:
        if consumer:
            unsubscribe(consumer);

view source code

Assignment subscription Offset

The assignment function is used to get the assignment of the topic.

try:
    assignments = consumer.assignment()
    if assignments:
        for partition in assignments:
            partition.offset = 0
            consumer.seek(partition)
            print(f"Assignment seek to beginning successfully.")
except Exception as err:
    print(f"Failed to seek offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
    raise err

view source code

The seek function is used to reset the assignment of the topic.

#!/usr/bin/python3
import taos

db              = "power"
topic           = "topic_meters"
user            = "root"
password        = "taosdata"
host            = "localhost"
port            = 6030
groupId         = "group1"
clientId        = "1"  
tdConnWsScheme  = "ws"
autoOffsetReset = "latest"
autoCommitState = "true"
autoCommitIntv  = "1000"


def prepareMeta():
    conn = None
    try:
        conn = taos.connect(host=host, user=user, password=password, port=port)
        conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")

        # change database. same as execute "USE db"
        conn.select_db(db)

        # create super table
        conn.execute(
            "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)"
        )

        # create topic
        conn.execute(
            f"CREATE TOPIC IF NOT EXISTS {topic} AS SELECT ts, current, voltage, phase, groupid, location FROM meters"
        )
        sql = """
            INSERT INTO 
            power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
                VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
                (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
            power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
                VALUES (NOW + 1a, 10.30000, 218, 0.25000)
            """
        affectedRows = conn.execute(sql)
        print(f"Inserted into {affectedRows} rows to power.meters successfully.")
    except Exception as err:
        print(f"Failed to prepareMeta, host: {host}:{port}, db: {db}, topic: {topic}, ErrMessage:{err}.")
        raise err
    finally:
        if conn:
            conn.close()

from taos.tmq import Consumer

def create_consumer():
    try:
        consumer = Consumer(
            {
                "group.id": groupId,
                "client.id": clientId,
                "td.connect.user": user,
                "td.connect.pass": password,
                "enable.auto.commit": autoCommitState,
                "auto.commit.interval.ms": autoCommitIntv,
                "auto.offset.reset": autoOffsetReset,
                "td.connect.ip": host,
                "td.connect.port": str(port),
            }
        )
        print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}")
        return consumer
    except Exception as err:
        print(f"Failed to create native consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err


def subscribe(consumer):
    try:
        # subscribe to the topics
        consumer.subscribe(["topic_meters"])
        print("Subscribe topics successfully")
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        data = block.fetchall()
                        print(f"data: {data}")

    except Exception as err:
        print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err



def commit_offset(consumer):
    try:
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        print(block.fetchall())

                # after processing the data, commit the offset manually
                consumer.commit(records)
                print("Commit offset manually successfully.");

    except Exception as err:
        print(f"Failed to commit offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err


def seek_offset(consumer):
    try:
        assignments = consumer.assignment()
        if assignments:
            for partition in assignments:
                partition.offset = 0
                consumer.seek(partition)
                print(f"Assignment seek to beginning successfully.")
    except Exception as err:
        print(f"Failed to seek offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err

def unsubscribe(consumer):
    try:
        consumer.unsubscribe()
        print("Consumer unsubscribed successfully.");
    except Exception as err:
        print(f"Failed to unsubscribe consumer. topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    finally:
        if consumer:
            consumer.close()
            print("Consumer closed successfully.");        

if __name__ == "__main__":
    consumer = None
    try:
        prepareMeta()
        consumer = create_consumer()
        subscribe(consumer)
        seek_offset(consumer)
        commit_offset(consumer)
    finally:
        if consumer:
            unsubscribe(consumer);

view source code

Close subscriptions

You should unsubscribe to the topics and close the consumer after consuming.

try:
    consumer.unsubscribe()
    print("Consumer unsubscribed successfully.");
except Exception as err:
    print(f"Failed to unsubscribe consumer. topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
    raise err
finally:
    if consumer:
        consumer.close()
        print("Consumer closed successfully.");        

view source code

Full Sample Code

#!/usr/bin/python3
import taos

db              = "power"
topic           = "topic_meters"
user            = "root"
password        = "taosdata"
host            = "localhost"
port            = 6030
groupId         = "group1"
clientId        = "1"  
tdConnWsScheme  = "ws"
autoOffsetReset = "latest"
autoCommitState = "true"
autoCommitIntv  = "1000"


def prepareMeta():
    conn = None
    try:
        conn = taos.connect(host=host, user=user, password=password, port=port)
        conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")

        # change database. same as execute "USE db"
        conn.select_db(db)

        # create super table
        conn.execute(
            "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)"
        )

        # ANCHOR: create_topic
        # create topic
        conn.execute(
            f"CREATE TOPIC IF NOT EXISTS {topic} AS SELECT ts, current, voltage, phase, groupid, location FROM meters"
        )
        # ANCHOR_END: create_topic
        sql = """
            INSERT INTO 
            power.d1001 USING power.meters (groupid, location) TAGS(2, 'California.SanFrancisco')
                VALUES (NOW + 1a, 10.30000, 219, 0.31000) 
                (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
            power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
                VALUES (NOW + 1a, 10.30000, 218, 0.25000)
            """
        affectedRows = conn.execute(sql)
        print(f"Inserted into {affectedRows} rows to power.meters successfully.")
    except Exception as err:
        print(f"Failed to prepareMeta, host: {host}:{port}, db: {db}, topic: {topic}, ErrMessage:{err}.")
        raise err
    finally:
        if conn:
            conn.close()

# ANCHOR: create_consumer
from taos.tmq import Consumer

def create_consumer():
    try:
        consumer = Consumer(
            {
                "group.id": groupId,
                "client.id": clientId,
                "td.connect.user": user,
                "td.connect.pass": password,
                "enable.auto.commit": autoCommitState,
                "auto.commit.interval.ms": autoCommitIntv,
                "auto.offset.reset": autoOffsetReset,
                "td.connect.ip": host,
                "td.connect.port": str(port),
            }
        )
        print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}")
        return consumer
    except Exception as err:
        print(f"Failed to create native consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    # ANCHOR_END: create_consumer


# ANCHOR: subscribe
def subscribe(consumer):
    try:
        # subscribe to the topics
        consumer.subscribe(["topic_meters"])
        print("Subscribe topics successfully")
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        data = block.fetchall()
                        print(f"data: {data}")

    except Exception as err:
        print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err


# ANCHOR_END: subscribe

def commit_offset(consumer):
    # ANCHOR: commit_offset
    try:
        for i in range(50):
            records = consumer.poll(1)
            if records:
                err = records.error()
                if err is not None:
                    print(f"Poll data error, {err}")
                    raise err

                val = records.value()
                if val:
                    for block in val:
                        print(block.fetchall())

                # after processing the data, commit the offset manually
                consumer.commit(records)
                print("Commit offset manually successfully.");

    except Exception as err:
        print(f"Failed to commit offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    # ANCHOR_END: commit_offset


def seek_offset(consumer):
    # ANCHOR: assignment
    try:
        assignments = consumer.assignment()
        if assignments:
            for partition in assignments:
                partition.offset = 0
                consumer.seek(partition)
                print(f"Assignment seek to beginning successfully.")
    except Exception as err:
        print(f"Failed to seek offset, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    # ANCHOR_END: assignment

def unsubscribe(consumer):
    # ANCHOR: unsubscribe    
    try:
        consumer.unsubscribe()
        print("Consumer unsubscribed successfully.");
    except Exception as err:
        print(f"Failed to unsubscribe consumer. topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
        raise err
    finally:
        if consumer:
            consumer.close()
            print("Consumer closed successfully.");        
    # ANCHOR_END: unsubscribe

if __name__ == "__main__":
    consumer = None
    try:
        prepareMeta()
        consumer = create_consumer()
        subscribe(consumer)
        seek_offset(consumer)
        commit_offset(consumer)
    finally:
        if consumer:
            unsubscribe(consumer);

view source code

Other sample programs

Example program links	Example program content
bind_multi.py	parameter binding, bind multiple rows at once
bind_row.py	parameter binding, bind one row at once
insert_lines.py	InfluxDB line protocol writing
json_tag.py	Use JSON type tags
tmq_consumer.py	TMQ subscription

Other notes

About nanoseconds

Due to the current imperfection of Python's nanosecond support (see link below), the current implementation returns integers at nanosecond precision instead of the datetime type produced by ms and us, which application developers will need to handle on their own. And it is recommended to use pandas' to_datetime(). The Python client library may modify the interface in the future if Python officially supports nanoseconds in full.

1. https://stackoverflow.com/questions/10611328/parsing-datetime-strings-containing-nanoseconds
2. https://www.python.org/dev/peps/pep-0564/

API Reference

• taos
• taosrest

Frequently Asked Questions

Welcome to ask questions or report questions.