Skip to main content

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 Versionmajor changes
2.7.121. added support for varbinary type (STMT does not yet support)
2. improved query performance (thanks to contributor hadrianl)
2.7.9support for getting assignment and seek function on subscription
2.7.8add execute_many method
Python Websocket Connection Versionmajor changes
0.2.9bugs fixes
0.2.51. support for getting assignment and seek function on subscription
2. support schemaless
3. support STMT
0.2.4support 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 TypeDescriptionSuggested Actions
InterfaceErrorthe native library is too old that it cannot support the functionplease check the TDengine client version
ConnectionErrorconnection errorplease check TDengine's status and the connection params
DatabaseErrordatabase errorplease upgrade Python client library to latest
OperationalErroroperation error
ProgrammingError
StatementErrorthe exception of stmt
ResultError
SchemalessErrorthe exception of stmt schemaless
TmqErrorthe 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 DataTypePython DataType
TIMESTAMPdatetime
INTint
BIGINTint
FLOATfloat
DOUBLEint
SMALLINTint
TINYINTint
BOOLbool
BINARYstr
NCHARstr
JSONstr

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 the latest version of:

pip3 install taospy

You can also specify a specific version to install:

pip3 install taospy==2.3.0

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

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

import taosrest
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.

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
}

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.

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.

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

conn = taosrest.connect(url="http://localhost:6041")
# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
conn.execute("DROP DATABASE IF EXISTS test")
conn.execute("CREATE DATABASE test")
conn.execute("USE test")
conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")

Insert data

conn.execute("INSERT INTO t1 USING weather TAGS(1) VALUES (now, 23.5) (now+1m, 23.5) (now+2m, 24.4)")

::: 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). :::

Basic Usage

Use of the RestClient class

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.

Use of RestClient
from taosrest import RestClient

client = RestClient("http://localhost:6041", user="root", password="taosdata")
res: dict = client.sql("SELECT ts, current FROM power.meters LIMIT 1")
print(res)

# output:
# {'status': 'succ', 'head': ['ts', 'current'], 'column_meta': [['ts', 9, 8], ['current', 6, 4]], 'data': [[datetime.datetime(2018, 10, 3, 14, 38, 5, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 10.3]], 'rows': 1}


view source code

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

Use of TaosRestCursor class

The TaosRestCursor class is an implementation of the PEP249 Cursor interface.

Use of TaosRestCursor
# create STable
cursor = conn.cursor()
cursor.execute("DROP DATABASE IF EXISTS power")
cursor.execute("CREATE DATABASE power keep 36500 ")
cursor.execute(
"CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)")

# insert data
cursor.execute("""INSERT INTO power.d1001 USING power.meters TAGS('California.SanFrancisco', 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000)
power.d1002 USING power.meters TAGS('California.SanFrancisco', 3) VALUES ('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000)
power.d1003 USING power.meters TAGS('California.LosAngeles', 2) VALUES ('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000) ('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000)
power.d1004 USING power.meters TAGS('California.LosAngeles', 3) VALUES ('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000) ('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)""")
print("inserted row count:", cursor.rowcount)

# query data
cursor.execute("SELECT * FROM power.meters LIMIT 3")
# get total rows
print("queried row count:", cursor.rowcount)
# get column names from cursor
column_names = [meta[0] for meta in cursor.description]
# get rows
data = cursor.fetchall()
print(column_names)
for row in data:
print(row)
# close cursor
cursor.close()

# output:
# inserted row count: 8
# queried row count: 3
# ['ts', 'current', 'voltage', 'phase', 'location', 'groupid']
# [datetime.datetime(2018, 10, 3, 14, 38, 5, 500000, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 11.8, 221, 0.28, 'california.losangeles', 2]
# [datetime.datetime(2018, 10, 3, 14, 38, 16, 600000, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 13.4, 223, 0.29, 'california.losangeles', 2]
# [datetime.datetime(2018, 10, 3, 14, 38, 5, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 10.8, 223, 0.29, 'california.losangeles', 3]

view source code

  • cursor.execute: Used to execute arbitrary SQL statements.
  • cursor.rowcount : For write operations, returns the number of successful rows written. For query operations, returns the number of rows in the result set.
  • cursor.description : Returns the description of the field. Please refer to TaosRestCursor for the specific format of the description information.
note

The best practice for TaosRestCursor is to create a cursor at the beginning of a query and close it immediately after use. Please avoid reusing the same cursor for multiple executions.

Querying Data

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.

from taosrest import RestClient

client = RestClient("http://localhost:6041", user="root", password="taosdata")
res: dict = client.sql("SELECT ts, current FROM power.meters LIMIT 1")
print(res)

# output:
# {'status': 'succ', 'head': ['ts', 'current'], 'column_meta': [['ts', 9, 8], ['current', 6, 4]], 'data': [[datetime.datetime(2018, 10, 3, 14, 38, 5, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 10.3]], 'rows': 1}


view source code

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

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.

Use of the RestClient class

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.

Use of RestClient
from taosrest import RestClient

client = RestClient("http://localhost:6041", user="root", password="taosdata")
res: dict = client.sql("SELECT ts, current FROM power.meters LIMIT 1", req_id=1)
print(res)

# output:
# {'status': 'succ', 'head': ['ts', 'current'], 'column_meta': [['ts', 9, 8], ['current', 6, 4]], 'data': [[datetime.datetime(2018, 10, 3, 14, 38, 5, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 10.3]], 'rows': 1}


view source code

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

Use of TaosRestCursor class

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

Use of TaosRestCursor
# create STable
cursor = conn.cursor()
cursor.execute("DROP DATABASE IF EXISTS power", req_id=1)
cursor.execute("CREATE DATABASE power keep 36500", req_id=2)
cursor.execute(
"CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)", req_id=3)

# insert data
cursor.execute("""INSERT INTO power.d1001 USING power.meters TAGS('California.SanFrancisco', 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000)
power.d1002 USING power.meters TAGS('California.SanFrancisco', 3) VALUES ('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000)
power.d1003 USING power.meters TAGS('California.LosAngeles', 2) VALUES ('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000) ('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000)
power.d1004 USING power.meters TAGS('California.LosAngeles', 3) VALUES ('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000) ('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)""", req_id=4)
print("inserted row count:", cursor.rowcount)

# query data
cursor.execute("SELECT * FROM power.meters LIMIT 3", req_id=5)
# get total rows
print("queried row count:", cursor.rowcount)
# get column names from cursor
column_names = [meta[0] for meta in cursor.description]
# get rows
data = cursor.fetchall()
print(column_names)
for row in data:
print(row)
# close cursor
cursor.close()

# output:
# inserted row count: 8
# queried row count: 3
# ['ts', 'current', 'voltage', 'phase', 'location', 'groupid']
# [datetime.datetime(2018, 10, 3, 14, 38, 5, 500000, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 11.8, 221, 0.28, 'california.losangeles', 2]
# [datetime.datetime(2018, 10, 3, 14, 38, 16, 600000, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 13.4, 223, 0.29, 'california.losangeles', 2]
# [datetime.datetime(2018, 10, 3, 14, 38, 5, tzinfo=datetime.timezone(datetime.timedelta(seconds=28800), '+08:00')), 10.8, 223, 0.29, 'california.losangeles', 3]

view source code

  • cursor.execute: Used to execute arbitrary SQL statements.
  • cursor.rowcount : For write operations, returns the number of successful rows written. For query operations, returns the number of rows in the result set.
  • cursor.description : Returns the description of the field. Please refer to TaosRestCursor for the specific format of the description information.

Used with pandas

import pandas
from sqlalchemy import create_engine, text

engine = create_engine("taosrest://root:taosdata@localhost:6041")
conn = engine.connect()
df: pandas.DataFrame = pandas.read_sql(text("SELECT * FROM power.meters"), conn)
conn.close()

# print index
print(df.index)
# print data type of element in ts column
print(type(df.ts[0]))
print(df.head(3))

# output:
# RangeIndex(start=0, stop=8, step=1)
# <class 'pandas._libs.tslibs.timestamps.Timestamp'>
# ts current ... location groupid
# 0 2018-10-03 06:38:05.500000+00:00 11.8 ... california.losangeles 2
# 1 2018-10-03 06:38:16.600000+00:00 13.4 ... california.losangeles 2
# 2 2018-10-03 06:38:05+00:00 10.8 ... california.losangeles 3

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.

Create Stmt

Call the statement method in Connection to create the stmt for parameter binding.

import taos

conn = taos.connect()
stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
parameter binding

Call the new_multi_binds function to create the parameter list for parameter bindings.

params = new_multi_binds(16)
params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
params[1].bool((True, None, False))
params[2].tinyint([-128, -128, None]) # -128 is tinyint null
params[3].tinyint([0, 127, None])
params[4].smallint([3, None, 2])
params[5].int([3, 4, None])
params[6].bigint([3, 4, None])
params[7].tinyint_unsigned([3, 4, None])
params[8].smallint_unsigned([3, 4, None])
params[9].int_unsigned([3, 4, None])
params[10].bigint_unsigned([3, 4, None])
params[11].float([3, None, 1])
params[12].double([3, None, 1.2])
params[13].binary(["abc", "dddafadfadfadfadfa", None])
params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
params[15].timestamp([None, None, 1626861392591])

Call the bind_param (for a single row) method or the bind_param_batch (for multiple rows) method to set the values.

stmt.bind_param_batch(params)
execute sql

Call execute method to execute sql.

stmt.execute()
Close Stmt
stmt.close()
Example
#!

import taosws

import taos

db_name = 'test_ws_stmt'


def before():
taos_conn = taos.connect()
taos_conn.execute("drop database if exists %s" % db_name)
taos_conn.execute("create database %s keep 36500" % db_name)
taos_conn.select_db(db_name)
taos_conn.execute("create table t1 (ts timestamp, a int, b float, c varchar(10))")
taos_conn.execute(
"create table stb1 (ts timestamp, a int, b float, c varchar(10)) tags (t1 int, t2 binary(10))")
taos_conn.close()


def stmt_insert():
before()

conn = taosws.connect('taosws://root:taosdata@localhost:6041/%s' % db_name)

while True:
try:
stmt = conn.statement()
stmt.prepare("insert into t1 values (?, ?, ?, ?)")

stmt.bind_param([
taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
taosws.ints_to_column([1, 2, 3, 4]),
taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
taosws.varchar_to_column(['a', 'b', 'c', 'd']),
])

stmt.add_batch()
rows = stmt.execute()
print(rows)
stmt.close()
except Exception as e:
if 'Retry needed' in e.args[0]: # deal with [0x0125] Retry needed
continue
else:
raise e

break


def stmt_insert_into_stable():
before()

conn = taosws.connect("taosws://root:taosdata@localhost:6041/%s" % db_name)

while True:
try:
stmt = conn.statement()
stmt.prepare("insert into ? using stb1 tags (?, ?) values (?, ?, ?, ?)")
stmt.set_tbname('stb1_1')
stmt.set_tags([
taosws.int_to_tag(1),
taosws.varchar_to_tag('aaa'),
])
stmt.bind_param([
taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
taosws.ints_to_column([1, 2, 3, 4]),
taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
taosws.varchar_to_column(['a', 'b', 'c', 'd']),
])

stmt.add_batch()
rows = stmt.execute()
print(rows)
stmt.close()
except Exception as e:
if 'Retry needed' in e.args[0]: # deal with [0x0125] Retry needed
continue
else:
raise e

break

view source code

Schemaless Writing

Client library support schemaless insert.

Simple insert
import taos

conn = taos.connect()
dbname = "pytest_line"
conn.execute("drop database if exists %s" % dbname)
conn.execute("create database if not exists %s precision 'us' keep 36500" % dbname)
conn.select_db(dbname)

lines = [
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000',
]
conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED)
print("inserted")

conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED)

result = conn.query("show tables")
for row in result:
print(row)

conn.execute("drop database if exists %s" % dbname)

view source code

Insert with ttl argument
import taos
from taos import SmlProtocol, SmlPrecision

conn = taos.connect()
dbname = "pytest_line"
conn.execute("drop database if exists %s" % dbname)
conn.execute("create database if not exists %s precision 'us' keep 36500" % dbname)
conn.select_db(dbname)

lines = [
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000',
]
conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, ttl=1000)
print("inserted")

conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, ttl=1000)

result = conn.query("show tables")
for row in result:
print(row)

conn.execute("drop database if exists %s" % dbname)

view source code

Insert with req_id argument
import taos
from taos import SmlProtocol, SmlPrecision

conn = taos.connect()
dbname = "pytest_line"
conn.execute("drop database if exists %s" % dbname)
conn.execute("create database if not exists %s precision 'us' keep 36500" % dbname)
conn.select_db(dbname)

lines = [
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000',
]
conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, req_id=1)
print("inserted")

conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, req_id=2)

result = conn.query("show tables")
for row in result:
print(row)

conn.execute("drop database if exists %s" % dbname)

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.

import taos
from taos import SmlProtocol, SmlPrecision

conn = taos.connect()
dbname = "pytest_line"
conn.execute("drop database if exists %s" % dbname)
conn.execute("create database if not exists %s precision 'us' keep 36500" % dbname)
conn.select_db(dbname)

lines = [
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000',
]
conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, req_id=1)
print("inserted")

conn.schemaless_insert(lines, taos.SmlProtocol.LINE_PROTOCOL, taos.SmlPrecision.NOT_CONFIGURED, req_id=2)

result = conn.query("show tables")
for row in result:
print(row)

conn.execute("drop database if exists %s" % dbname)

view source code

import taos
from taos import utils
from taos import TaosConnection
from taos.cinterface import *
from taos.error import OperationalError, SchemalessError

conn = taos.connect()
dbname = "taos_schemaless_insert"
try:
conn.execute("drop database if exists %s" % dbname)

if taos.IS_V3:
conn.execute("create database if not exists %s schemaless 1 precision 'ns' keep 36500" % dbname)
else:
conn.execute("create database if not exists %s update 2 precision 'ns' keep 36500" % dbname)

conn.select_db(dbname)

lines = '''st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"passit",c2=false,c4=4f64 1626006833639000000
st,t1=4i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"passitagin, abc",c2=true,c4=5f64,c5=5f64,c6=7u64 1626006933640000000
stf,t1=4i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"passitagin_stf",c2=false,c5=5f64,c6=7u64 1626006933641000000'''

ttl = 1000
req_id = utils.gen_req_id()
res = conn.schemaless_insert_raw(lines, 1, 0, ttl=ttl, req_id=req_id)
print("affected rows: ", res)
assert (res == 3)

lines = '''stf,t1=5i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"passitagin_stf",c2=false,c5=5f64,c6=7u64 1626006933641000000'''
ttl = 1000
req_id = utils.gen_req_id()
res = conn.schemaless_insert_raw(lines, 1, 0, ttl=ttl, req_id=req_id)
print("affected rows: ", res)
assert (res == 1)

result = conn.query("select * from st")
dict2 = result.fetch_all_into_dict()
print(dict2)
print(result.row_count)

all = result.rows_iter()
for row in all:
print(row)
result.close()
assert (result.row_count == 2)

# error test
lines = ''',t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"passit",c2=false,c4=4f64 1626006833639000000'''
try:
ttl = 1000
req_id = utils.gen_req_id()
res = conn.schemaless_insert_raw(lines, 1, 0, ttl=ttl, req_id=req_id)
print(res)
# assert(False)
except SchemalessError as err:
print('**** error: ', err)
# assert (err.msg == 'Invalid data format')

result = conn.query("select * from st")
print(result.row_count)
all = result.rows_iter()
for row in all:
print(row)
result.close()

conn.execute("drop database if exists %s" % dbname)
conn.close()
except InterfaceError as err:
conn.execute("drop database if exists %s" % dbname)
conn.close()
print(err)
except Exception as err:
conn.execute("drop database if exists %s" % dbname)
conn.close()
print(err)
raise err

view source code

Data Subscription

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

Create a Topic

To create topic, please refer to Data Subscription.

Create a Consumer

The consumer in the client library contains the subscription api. The syntax for creating a consumer is consumer = Consumer(configs). For more subscription api parameters, please refer to Data Subscription.

from taos.tmq import Consumer

consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})

Subscribe to a Topic

The subscribe function is used to subscribe to a list of topics.

consumer.subscribe(['topic1', 'topic2'])

Consume messages

The poll function is used to consume data in tmq. The parameter of the poll function is a value of type float representing the timeout in seconds. It returns a Message before timing out, or None on timing out. You have to handle error messages in response data.

while True:
res = consumer.poll(1)
if not res:
continue
err = res.error()
if err is not None:
raise err
val = res.value()

for block in val:
print(block.fetchall())

Assignment subscription Offset

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

assignments = consumer.assignment()

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

tp = TopicPartition(topic='topic1', partition=0, offset=0)
consumer.seek(tp)

Close subscriptions

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

consumer.unsubscribe()
consumer.close()

Full Sample Code

from taos.tmq import Consumer
import taos


def init_tmq_env(db, topic):
conn = taos.connect()
conn.execute("drop topic if exists {}".format(topic))
conn.execute("drop database if exists {}".format(db))
conn.execute("create database if not exists {} wal_retention_period 3600 keep 36500".format(db))
conn.select_db(db)
conn.execute(
"create stable if not exists stb1 (ts timestamp, c1 int, c2 float, c3 varchar(16)) tags(t1 int, t3 varchar(16))")
conn.execute("create table if not exists tb1 using stb1 tags(1, 't1')")
conn.execute("create table if not exists tb2 using stb1 tags(2, 't2')")
conn.execute("create table if not exists tb3 using stb1 tags(3, 't3')")
conn.execute("create topic if not exists {} as select ts, c1, c2, c3 from stb1".format(topic))
conn.execute("insert into tb1 values (now, 1, 1.0, 'tmq test')")
conn.execute("insert into tb2 values (now, 2, 2.0, 'tmq test')")
conn.execute("insert into tb3 values (now, 3, 3.0, 'tmq test')")


def cleanup(db, topic):
conn = taos.connect()
conn.execute("drop topic if exists {}".format(topic))
conn.execute("drop database if exists {}".format(db))


if __name__ == '__main__':
init_tmq_env("tmq_test", "tmq_test_topic") # init env
consumer = Consumer(
{
"group.id": "tg2",
"td.connect.user": "root",
"td.connect.pass": "taosdata",
"enable.auto.commit": "true",
}
)
consumer.subscribe(["tmq_test_topic"])

try:
while True:
res = consumer.poll(1)
if not res:
break
err = res.error()
if err is not None:
raise err
val = res.value()

for block in val:
print(block.fetchall())
finally:
consumer.unsubscribe()
consumer.close()
cleanup("tmq_test", "tmq_test_topic")

view source code

import taos
from taos.tmq import Consumer
import taosws


def prepare():
conn = taos.connect()
conn.execute("drop topic if exists tmq_assignment_demo_topic")
conn.execute("drop database if exists tmq_assignment_demo_db")
conn.execute("create database if not exists tmq_assignment_demo_db wal_retention_period 3600 keep 36500")
conn.select_db("tmq_assignment_demo_db")
conn.execute(
"create table if not exists tmq_assignment_demo_table (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)")
conn.execute(
"create topic if not exists tmq_assignment_demo_topic as select ts, c1, c2, c3 from tmq_assignment_demo_table")
conn.execute("insert into d0 using tmq_assignment_demo_table tags (0) values (now-2s, 1, 1.0, 'tmq test')")
conn.execute("insert into d0 using tmq_assignment_demo_table tags (0) values (now-1s, 2, 2.0, 'tmq test')")
conn.execute("insert into d0 using tmq_assignment_demo_table tags (0) values (now, 3, 3.0, 'tmq test')")


def taos_get_assignment_and_seek_demo():
prepare()
consumer = Consumer(
{
"group.id": "0",
}
)
consumer.subscribe(["tmq_assignment_demo_topic"])

# get topic assignment
assignments = consumer.assignment()
for assignment in assignments:
print(assignment)

# poll
consumer.poll(1)
consumer.poll(1)

# get topic assignment again
after_pool_assignments = consumer.assignment()
for assignment in after_pool_assignments:
print(assignment)

# seek to the beginning
for assignment in assignments:
consumer.seek(assignment)

# now the assignment should be the same as before poll
assignments = consumer.assignment()
for assignment in assignments:
print(assignment)


if __name__ == '__main__':
taos_get_assignment_and_seek_demo()

view source code

Other sample programs

Example program linksExample program content
bind_multi.pyparameter binding, bind multiple rows at once
bind_row.pyparameter binding, bind one row at once
insert_lines.pyInfluxDB line protocol writing
json_tag.pyUse JSON type tags
tmq_consumer.pyTMQ 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

Frequently Asked Questions

Welcome to ask questions or report questions.