Skip to content

Commit 2fe2bb8

Browse files
qevolgqevolg
authored
feat: tmq supports token (#385)
close: https://project.feishu.cn/taosdata_td/feature/detail/6735486234
1 parent 1d4a6d5 commit 2fe2bb8

14 files changed

Lines changed: 234 additions & 37 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
77

88
## [Unreleased]
99

10+
## v2.8.9 - 2026-03-05
11+
12+
### Features:
13+
14+
- Data subscription supports token authentication
15+
1016
## v2.8.6 - 2025-10-27
1117

1218
### Features:

‎ci/release-ws.sh‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,6 @@ sed "1,9d" CHANGELOG.md >> CHANGELOG.md2
1717
mv CHANGELOG.md2 CHANGELOG.md
1818

1919
# git commit -a -m "release(taos-ws-py): v$newv"
20-
#git push
21-
#git tag taos-ws-py-v$newv
22-
#git push --force origin taos-ws-py-v$newv:taos-ws-py-v$newv
20+
# git push
21+
# git tag taos-ws-py-v$newv
22+
# git push --force origin taos-ws-py-v$newv:taos-ws-py-v$newv

‎ci/release.sh‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@ $ci/changelog-generate.sh >> CHANGELOG.md2
1515
sed "1,9d" CHANGELOG.md >> CHANGELOG.md2
1616
mv CHANGELOG.md2 CHANGELOG.md
1717

18-
git commit -a -m "release: v$newv"
19-
git push
20-
git tag v$newv
21-
git push --force origin v$newv:v$newv
22-
#bash build_doc.sh
18+
# git commit -a -m "release: v$newv"
19+
# git push
20+
# git tag v$newv
21+
# git push --force origin v$newv:v$newv
22+
# bash build_doc.sh

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[tool.poetry]
22
name = "taospy"
3-
version = "2.8.6"
3+
version = "2.8.9"
44
description = "The official TDengine Python connector"
55
authors = ["Taosdata Inc. <support@taosdata.com>"]
66
license = "MIT"

‎taos-ws-py/CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
55
The format is based on [Conventional Changelog](https://www.conventionalcommits.org/en/v1.0.0/),
66
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
77

8+
## v0.6.6 - 2026-03-05
9+
10+
### Features:
11+
12+
- Data subscription supports token authentication
13+
814
## v0.6.5 - 2025-12-31
915

1016
### Features:

‎taos-ws-py/Cargo.lock‎

Lines changed: 12 additions & 12 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎taos-ws-py/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "taos-ws-py"
3-
version = "0.6.5"
3+
version = "0.6.6"
44
edition = "2021"
55
publish = false
66
license = "MIT"

‎taos-ws-py/src/consumer.rs‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,9 @@ impl Consumer {
5858
}
5959

6060
let mut tz = dsn_cfg.get("timezone").cloned();
61+
if let Some(token) = dsn_cfg.remove("td.connect.bearer_token") {
62+
dsn_cfg.set("bearer_token", token);
63+
}
6164

6265
if let Some(args) = conf {
6366
if let Some(scheme) = args
@@ -105,6 +108,12 @@ impl Consumer {
105108
if let Some(value) = args.get_item("td.connect.token").or(args.get_item("token")) {
106109
dsn_cfg.set("token", value.extract::<String>()?);
107110
}
111+
if let Some(value) = args
112+
.get_item("td.connect.bearer_token")
113+
.or(args.get_item("bearer_token"))
114+
{
115+
dsn_cfg.set("bearer_token", value.extract::<String>()?);
116+
}
108117
if let Some(value) = args.get_item("timezone") {
109118
tz = Some(value.extract::<String>()?);
110119
}
@@ -135,6 +144,8 @@ impl Consumer {
135144
"td.connect.user",
136145
"td.connect.pass",
137146
"group.id",
147+
"bearer_token",
148+
"td.connect.bearer_token",
138149
];
139150

140151
for (key, value) in args {

‎taos-ws-py/tests/test_tmq.py‎

Lines changed: 124 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
1+
import pytest
12
import taosws
3+
import time
4+
import utils
25
from taosws import Consumer
36

47

@@ -55,7 +58,6 @@ def setup():
5558
"alter table `table` drop column new1",
5659
]
5760
for statement in statements:
58-
# print(statement)
5961
cursor.execute(statement)
6062

6163

@@ -69,7 +71,6 @@ def test_tmq():
6971
"min.poll.rows": "129",
7072
}
7173
consumer = Consumer(conf)
72-
7374
consumer.subscribe(["ws_tmq_meta"])
7475

7576
while 1:
@@ -95,6 +96,127 @@ def test_tmq():
9596
consumer.unsubscribe()
9697

9798

99+
@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
100+
def test_tmq_with_token():
101+
conn = taosws.connect("ws://localhost:6041")
102+
try:
103+
conn.execute("drop token if exists token_1772607422")
104+
conn.execute("drop topic if exists topic_1772607422")
105+
conn.execute("drop database if exists test_1772607422")
106+
conn.execute("create database test_1772607422")
107+
conn.execute("create topic topic_1772607422 as database test_1772607422")
108+
109+
rs = conn.query(f"create token token_1772607422 from user {utils.test_username()}")
110+
token = next(iter(rs))[0]
111+
112+
consumer = Consumer(conf={
113+
"td.connect.websocket.scheme": "ws",
114+
"td.connect.ip": "localhost",
115+
"td.connect.port": 6041,
116+
"td.connect.user": "invalid_user",
117+
"td.connect.pass": "invalid_pass",
118+
"td.connect.bearer_token": token,
119+
"group.id": "1001",
120+
"client.id": "1001",
121+
})
122+
consumer.subscribe(["topic_1772607422"])
123+
consumer.unsubscribe()
124+
125+
consumer2 = Consumer(conf={
126+
"td.connect.websocket.scheme": "ws",
127+
"td.connect.ip": "localhost",
128+
"td.connect.port": 6041,
129+
"td.connect.user": "invalid_user",
130+
"td.connect.pass": "invalid_pass",
131+
"bearer_token": token,
132+
"group.id": "1001",
133+
"client.id": "1001",
134+
})
135+
consumer2.subscribe(["topic_1772607422"])
136+
consumer2.unsubscribe()
137+
138+
consumer3 = Consumer(conf={
139+
"td.connect.websocket.scheme": "ws",
140+
"td.connect.ip": "localhost",
141+
"td.connect.port": 6041,
142+
"td.connect.user": "invalid_user",
143+
"td.connect.pass": "invalid_pass",
144+
"td.connect.bearer_token": token,
145+
"bearer_token": "invalid_token",
146+
"group.id": "1001",
147+
"client.id": "1001",
148+
})
149+
consumer3.subscribe(["topic_1772607422"])
150+
consumer3.unsubscribe()
151+
152+
consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token={token}")
153+
consumer4.subscribe(["topic_1772607422"])
154+
consumer4.unsubscribe()
155+
156+
consumer5 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token={token}")
157+
consumer5.subscribe(["topic_1772607422"])
158+
consumer5.unsubscribe()
159+
160+
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")
161+
consumer6.subscribe(["topic_1772607422"])
162+
consumer6.unsubscribe()
163+
finally:
164+
time.sleep(3)
165+
conn.execute("drop token if exists token_1772607422")
166+
conn.execute("drop topic if exists topic_1772607422")
167+
conn.execute("drop database if exists test_1772607422")
168+
conn.close()
169+
170+
171+
@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
172+
def test_tmq_with_invalid_token():
173+
conn = taosws.connect("ws://localhost:6041")
174+
try:
175+
conn.execute("drop topic if exists topic_1772611547")
176+
conn.execute("drop database if exists test_1772611547")
177+
conn.execute("create database test_1772611547")
178+
conn.execute("create topic topic_1772611547 as database test_1772611547")
179+
180+
consumer = Consumer(conf={
181+
"td.connect.websocket.scheme": "ws",
182+
"td.connect.ip": "localhost",
183+
"td.connect.port": 6041,
184+
"td.connect.user": "invalid_user",
185+
"td.connect.pass": "invalid_pass",
186+
"td.connect.bearer_token": "invalid_token",
187+
"group.id": "1001",
188+
"client.id": "1001",
189+
})
190+
with pytest.raises(Exception, match=r"init tscObj with token failed"):
191+
consumer.subscribe(["topic_1772611547"])
192+
193+
consumer2 = Consumer(conf={
194+
"td.connect.websocket.scheme": "ws",
195+
"td.connect.ip": "localhost",
196+
"td.connect.port": 6041,
197+
"td.connect.user": "invalid_user",
198+
"td.connect.pass": "invalid_pass",
199+
"bearer_token": "invalid_token",
200+
"group.id": "1001",
201+
"client.id": "1001",
202+
})
203+
with pytest.raises(Exception, match=r"init tscObj with token failed"):
204+
consumer2.subscribe(["topic_1772611547"])
205+
206+
consumer3 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token=invalid_token")
207+
with pytest.raises(Exception, match=r"init tscObj with token failed"):
208+
consumer3.subscribe(["topic_1772611547"])
209+
210+
consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token=invalid_token")
211+
with pytest.raises(Exception, match=r"init tscObj with token failed"):
212+
consumer4.subscribe(["topic_1772611547"])
213+
finally:
214+
time.sleep(3)
215+
conn.execute("drop topic if exists topic_1772611547")
216+
conn.execute("drop database if exists test_1772611547")
217+
conn.close()
218+
219+
98220
def show_env():
99221
import os
100222

‎taos/_version.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
__version__ = "2.8.6"
1+
__version__ = '2.8.9'

0 commit comments

Comments
 (0)