TDengine Python 连接器 — taospy 与 taos-ws-py 完整指南

# Native (需要 libtaos)pip install taospy# WebSocket (推荐)pip install taos-ws-py# 两个都装（按需选用）pip install taospy taos-ws-py

# Nativeimport taosconn = taos.connect(    host="localhost",    port=6030,    user="root",    password="taosdata",    database="test")# WebSocketimport taoswsconn = taosws.connect("taosws://root:taosdata@localhost:6041/test")# 查询cursor = conn.cursor()cursor.execute("CREATE DATABASE IF NOT EXISTS test")cursor.execute("USE test")cursor.execute("""  CREATE STABLE IF NOT EXISTS meters (    ts TIMESTAMP, current FLOAT, voltage INT  ) TAGS (location VARCHAR(32))""")# 插入cursor.execute("""  INSERT INTO d001 USING meters TAGS('Beijing')   VALUES (NOW, 25.3, 220)""")# 查询cursor.execute("SELECT * FROM meters LIMIT 10")for row in cursor:    print(row)cursor.close()conn.close()

# 标准 DB-APIcursor.execute(    "SELECT * FROM meters WHERE ts > ? AND current > ?",    (datetime.now() - timedelta(hours=1), 100))# 批量插入cursor.executemany(    "INSERT INTO d001 VALUES (?, ?, ?)",    [(ts1, 25.0, 220), (ts2, 26.0, 221), ...])

# WebSocket STMT2import taoswsconn = taosws.connect("taosws://root:taosdata@localhost:6041/test")stmt = conn.statement2(    "INSERT INTO ? USING meters TAGS(?) VALUES(?, ?, ?)")# 准备多个子表的数据table_names = ["d001", "d002"]tags = [["Beijing"], ["Shanghai"]]cols = [    [        [int(time.time()*1000), int(time.time()*1000)+1000],  # ts        [25.3, 26.1],                                          # current        [220, 221]                                             # voltage    ],    [        [int(time.time()*1000), int(time.time()*1000)+1000],        [27.5, 28.2],        [218, 219]    ]]stmt.bind_param(table_names, tags, cols)stmt.execute()stmt.close()

import taosfrom taos import SmlProtocol, SmlPrecisionconn = taos.connect(host="localhost", database="test")# InfluxDB Linelines = [    "meters,location=Beijing current=25.3,voltage=220i 1717488000000",    "meters,location=Shanghai current=26.1,voltage=219i 1717488001000"]conn.schemaless_insert(    lines,     SmlProtocol.LINE_PROTOCOL,    SmlPrecision.MILLI_SECONDS)# OpenTSDB JSONimport jsondata = [{    "metric": "meters.current",    "timestamp": 1717488000,    "value": 25.3,    "tags": {"location": "Beijing"}}]conn.schemaless_insert(    [json.dumps(data)],    SmlProtocol.JSON_PROTOCOL,    SmlPrecision.SECONDS)

from taos.tmq import Consumerconsumer = Consumer({    "td.connect.ip": "localhost",    "td.connect.user": "root",    "td.connect.pass": "taosdata",    "group.id": "my_group",    "auto.offset.reset": "earliest",    "enable.auto.commit": "true"})consumer.subscribe(["my_topic"])try:    while True:        msg = consumer.poll(1)  # 1 秒超时        if msg is None:            continue        for block in msg:            for row in block:                print(row)        consumer.commit()except KeyboardInterrupt:    passfinally:    consumer.close()# WebSocket 版本from taosws import TmqConsumerconsumer = TmqConsumer(    "taosws://root:taosdata@localhost:6041/",    group_id="my_group",    auto_offset_reset="earliest")consumer.subscribe(["my_topic"])

import pandas as pdimport taoswsconn = taosws.connect("taosws://root:taosdata@localhost:6041/test")# 查询 → DataFramedf = pd.read_sql(    "SELECT _wstart, AVG(current) AS avg_c FROM meters "    "WHERE ts > NOW - 1d INTERVAL(1h)",    conn)print(df.head())# 处理df.set_index('_wstart', inplace=True)df.plot(figsize=(12, 4))# DataFrame → 写入df_to_insert = pd.DataFrame({    'ts': pd.date_range('2026-06-04', periods=1000, freq='S'),    'current': np.random.rand(1000) * 30,    'voltage': np.random.randint(218, 223, 1000)})cursor = conn.cursor()for _, row in df_to_insert.iterrows():    cursor.execute(        f"INSERT INTO d001 VALUES ('{row.ts}', {row.current}, {row.voltage})"    )

from sqlalchemy import create_engine, text# Nativeengine = create_engine("taos://root:taosdata@localhost:6030/test")# WebSocketengine = create_engine("taosws://root:taosdata@localhost:6041/test")# RESTengine = create_engine("taosrest://root:taosdata@localhost:6041/test")with engine.connect() as conn:    result = conn.execute(text("SELECT * FROM meters LIMIT 10"))    for row in result:        print(row)# Pandas + SQLAlchemydf = pd.read_sql("SELECT * FROM meters LIMIT 100", engine)

import timeimport randomimport taoswsfrom datetime import datetimedef main():    conn = taosws.connect(        "taosws://root:taosdata@localhost:6041/iot")    cursor = conn.cursor()    # 准备 schema    cursor.execute("""      CREATE DATABASE IF NOT EXISTS iot    """)    cursor.execute("""      CREATE STABLE IF NOT EXISTS sensors (        ts TIMESTAMP, temp FLOAT, humidity FLOAT      ) TAGS (device_id VARCHAR(32), location VARCHAR(32))    """)    # 模拟采集    devices = [        ("d001", "Floor1"),        ("d002", "Floor1"),        ("d003", "Floor2"),    ]    while True:        ts = int(time.time() * 1000)        for dev_id, loc in devices:            temp = 20 + random.random() * 10            humidity = 40 + random.random() * 30            cursor.execute(                f"INSERT INTO {dev_id} USING sensors "                f"TAGS('{dev_id}', '{loc}') "                f"VALUES ({ts}, {temp:.2f}, {humidity:.2f})"            )        time.sleep(1)if __name__ == "__main__":    main()

import taoswsimport pandas as pdimport matplotlib.pyplot as pltconn = taosws.connect("taosws://root:taosdata@localhost:6041/iot")# 小时聚合df = pd.read_sql("""  SELECT _wstart, location, AVG(temp) AS avg_temp   FROM sensors   WHERE ts > NOW - 1d   PARTITION BY location   INTERVAL(1h)""", conn)# 数据透视pivot = df.pivot(index='_wstart', columns='location', values='avg_temp')pivot.plot(figsize=(12, 6))plt.title("Hourly Temperature by Location")plt.savefig("temp_trend.png")

# 下载 taosToolswget ... && tar xzf ...sudo ./install.sh -e no

import taoswsasync def query():    conn = await taosws.connect_async("...")    cursor = await conn.cursor_async()    await cursor.execute("SELECT ...")

conn = taos.connect(..., config="/etc/taos")# taos.cfg 中设 locale en_US.UTF-8