feat: tmq supports token

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## v2.8.9 - 2026-03-05
+
+### Features:
+
+- Data subscription supports token authentication
+
 ## v2.8.6 - 2025-10-27
 
 ### Features:

ci/release-ws.sh

@@ -17,6 +17,6 @@ sed "1,9d" CHANGELOG.md >> CHANGELOG.md2
 mv CHANGELOG.md2 CHANGELOG.md
 
 # git commit -a -m "release(taos-ws-py): v$newv"
-#git push
-#git tag taos-ws-py-v$newv
-#git push --force origin taos-ws-py-v$newv:taos-ws-py-v$newv
+# git push
+# git tag taos-ws-py-v$newv
+# git push --force origin taos-ws-py-v$newv:taos-ws-py-v$newv

ci/release.sh

@@ -15,8 +15,8 @@ $ci/changelog-generate.sh >> CHANGELOG.md2
 sed "1,9d" CHANGELOG.md >> CHANGELOG.md2
 mv CHANGELOG.md2 CHANGELOG.md
 
-git commit -a -m "release: v$newv"
-git push
-git tag v$newv
-git push --force origin v$newv:v$newv
-#bash build_doc.sh
+# git commit -a -m "release: v$newv"
+# git push
+# git tag v$newv
+# git push --force origin v$newv:v$newv
+# bash build_doc.sh

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taospy"
-version = "2.8.6"
+version = "2.8.9"
 description = "The official TDengine Python connector"
 authors = ["Taosdata Inc. <support@taosdata.com>"]
 license = "MIT"

taos-ws-py/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Conventional Changelog](https://www.conventionalcommits.org/en/v1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## v0.6.6 - 2026-03-05
+
+### Features:
+
+- Data subscription supports token authentication
+
 ## v0.6.5 - 2025-12-31
 
 ### Features:

taos-ws-py/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "taos-ws-py"
-version = "0.6.5"
+version = "0.6.6"
 edition = "2021"
 publish = false
 license = "MIT"

taos-ws-py/src/consumer.rs

@@ -58,6 +58,9 @@ impl Consumer {
         }
 
         let mut tz = dsn_cfg.get("timezone").cloned();
+        if let Some(token) = dsn_cfg.remove("td.connect.bearer_token") {
+            dsn_cfg.set("bearer_token", token);
+        }
 
         if let Some(args) = conf {
             if let Some(scheme) = args
@@ -105,6 +108,12 @@ impl Consumer {
             if let Some(value) = args.get_item("td.connect.token").or(args.get_item("token")) {
                 dsn_cfg.set("token", value.extract::<String>()?);
             }
+            if let Some(value) = args
+                .get_item("td.connect.bearer_token")
+                .or(args.get_item("bearer_token"))
+            {
+                dsn_cfg.set("bearer_token", value.extract::<String>()?);
+            }
             if let Some(value) = args.get_item("timezone") {
                 tz = Some(value.extract::<String>()?);
             }
@@ -135,6 +144,8 @@ impl Consumer {
                 "td.connect.user",
                 "td.connect.pass",
                 "group.id",
+                "bearer_token",
+                "td.connect.bearer_token",
             ];
 
             for (key, value) in args {

taos-ws-py/tests/test_tmq.py

@@ -1,4 +1,7 @@
+import pytest
 import taosws
+import time
+import utils
 from taosws import Consumer
 
 
@@ -55,7 +58,6 @@ def setup():
         "alter table `table` drop column new1",
     ]
     for statement in statements:
-        # print(statement)
         cursor.execute(statement)
 
 
@@ -69,7 +71,6 @@ def test_tmq():
         "min.poll.rows": "129",
     }
     consumer = Consumer(conf)
-
     consumer.subscribe(["ws_tmq_meta"])
 
     while 1:
@@ -95,6 +96,127 @@ def test_tmq():
     consumer.unsubscribe()
 
 
+@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
+def test_tmq_with_token():
+    conn = taosws.connect("ws://localhost:6041")
+    try:
+        conn.execute("drop token if exists token_1772607422")
+        conn.execute("drop topic if exists topic_1772607422")
+        conn.execute("drop database if exists test_1772607422")
+        conn.execute("create database test_1772607422")
+        conn.execute("create topic topic_1772607422 as database test_1772607422")
+
+        rs = conn.query(f"create token token_1772607422 from user {utils.test_username()}")
+        token = next(iter(rs))[0]
+
+        consumer = Consumer(conf={
+            "td.connect.websocket.scheme": "ws",
+            "td.connect.ip": "localhost",
+            "td.connect.port": 6041,
+            "td.connect.user": "invalid_user",
+            "td.connect.pass": "invalid_pass",
+            "td.connect.bearer_token": token,
+            "group.id": "1001",
+            "client.id": "1001",
+        })
+        consumer.subscribe(["topic_1772607422"])
+        consumer.unsubscribe()
+
+        consumer2 = Consumer(conf={
+            "td.connect.websocket.scheme": "ws",
+            "td.connect.ip": "localhost",
+            "td.connect.port": 6041,
+            "td.connect.user": "invalid_user",
+            "td.connect.pass": "invalid_pass",
+            "bearer_token": token,
+            "group.id": "1001",
+            "client.id": "1001",
+        })
+        consumer2.subscribe(["topic_1772607422"])
+        consumer2.unsubscribe()
+
+        consumer3 = Consumer(conf={
+            "td.connect.websocket.scheme": "ws",
+            "td.connect.ip": "localhost",
+            "td.connect.port": 6041,
+            "td.connect.user": "invalid_user",
+            "td.connect.pass": "invalid_pass",
+            "td.connect.bearer_token": token,
+            "bearer_token": "invalid_token",
+            "group.id": "1001",
+            "client.id": "1001",
+        })
+        consumer3.subscribe(["topic_1772607422"])
+        consumer3.unsubscribe()
+
+        consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token={token}")
+        consumer4.subscribe(["topic_1772607422"])
+        consumer4.unsubscribe()
+
+        consumer5 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token={token}")
+        consumer5.subscribe(["topic_1772607422"])
+        consumer5.unsubscribe()
+
+        consumer6 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token={token}&bearer_token=invalid_token")
+        consumer6.subscribe(["topic_1772607422"])
+        consumer6.unsubscribe()
+    finally:
+        time.sleep(3)
+        conn.execute("drop token if exists token_1772607422")
+        conn.execute("drop topic if exists topic_1772607422")
+        conn.execute("drop database if exists test_1772607422")
+        conn.close()
+
+
+@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
+def test_tmq_with_invalid_token():
+    conn = taosws.connect("ws://localhost:6041")
+    try:
+        conn.execute("drop topic if exists topic_1772611547")
+        conn.execute("drop database if exists test_1772611547")
+        conn.execute("create database test_1772611547")
+        conn.execute("create topic topic_1772611547 as database test_1772611547")
+
+        consumer = Consumer(conf={
+            "td.connect.websocket.scheme": "ws",
+            "td.connect.ip": "localhost",
+            "td.connect.port": 6041,
+            "td.connect.user": "invalid_user",
+            "td.connect.pass": "invalid_pass",
+            "td.connect.bearer_token": "invalid_token",
+            "group.id": "1001",
+            "client.id": "1001",
+        })
+        with pytest.raises(Exception, match=r"init tscObj with token failed"):
+            consumer.subscribe(["topic_1772611547"])
+
+        consumer2 = Consumer(conf={
+            "td.connect.websocket.scheme": "ws",
+            "td.connect.ip": "localhost",
+            "td.connect.port": 6041,
+            "td.connect.user": "invalid_user",
+            "td.connect.pass": "invalid_pass",
+            "bearer_token": "invalid_token",
+            "group.id": "1001",
+            "client.id": "1001",
+        })
+        with pytest.raises(Exception, match=r"init tscObj with token failed"):
+            consumer2.subscribe(["topic_1772611547"])
+
+        consumer3 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token=invalid_token")
+        with pytest.raises(Exception, match=r"init tscObj with token failed"):
+            consumer3.subscribe(["topic_1772611547"])
+
+        consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token=invalid_token")
+        with pytest.raises(Exception, match=r"init tscObj with token failed"):
+            consumer4.subscribe(["topic_1772611547"])
+    finally:
+        time.sleep(3)
+        conn.execute("drop topic if exists topic_1772611547")
+        conn.execute("drop database if exists test_1772611547")
+        conn.close()
+
+
 def show_env():
     import os
 

taos/_version.py

@@ -1 +1 @@
-__version__ = "2.8.6"
+__version__ = '2.8.9'

taos/cinterface.py

@@ -1955,13 +1955,14 @@ def tmq_conf_set_auto_commit_cb(conf, cb, param):
     _UNSUPPORTED["tmq_consumer_new"] = err
 
 
-def tmq_consumer_new(conf, errstrlen=0):
-    # type (c_void_p, c_char_p, c_int) -> c_void_p
+def tmq_consumer_new(conf):
+    # type: (c_void_p) -> c_void_p
     _check_if_supported()
-    buf = ctypes.create_string_buffer(errstrlen)
-    tmq = cast(_libtaos.tmq_consumer_new(conf, buf, errstrlen), c_void_p)
+    errstr = ctypes.create_string_buffer(256)
+    tmq = cast(_libtaos.tmq_consumer_new(conf, errstr, 255), c_void_p)
     if tmq.value is None:
-        raise TmqError("failed on tmq_consumer_new()")
+        err = errstr.value.decode("utf-8", errors="replace")
+        raise TmqError(f"failed on tmq_consumer_new(), err: {err}")
     return tmq
 
 

taos/error.py

@@ -99,10 +99,6 @@ def __init__(self, msg=None, errno=0xFFFF, affected_rows=0):
     def __str__(self):
         return self._full_msg + "(affected rows: %d)" % self.affected_rows
 
-    # @property
-    # def affected_rows(self):
-    #     return self.affected_rows
-
 
 class LinesError(DatabaseError):
     """taos_insert_lines errors."""