Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
33f35846
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
未验证
提交
33f35846
编写于
12月 23, 2021
作者:
L
Linhe Huo
提交者:
GitHub
12月 23, 2021
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[TD-12250]<docs>: fix confused titles and syntax error of example code in python connector (#9252)
上级
47d563e1
变更
7
显示空白变更内容
内联
并排
Showing
7 changed file
with
131 addition
and
53 deletion
+131
-53
src/connector/python/README.md
src/connector/python/README.md
+59
-40
src/connector/python/examples/query-async.py
src/connector/python/examples/query-async.py
+1
-1
src/connector/python/examples/stream.py
src/connector/python/examples/stream.py
+59
-0
src/connector/python/taos/__init__.py
src/connector/python/taos/__init__.py
+2
-2
src/connector/python/taos/bind.py
src/connector/python/taos/bind.py
+1
-1
src/connector/python/taos/cinterface.py
src/connector/python/taos/cinterface.py
+4
-4
src/connector/python/taos/result.py
src/connector/python/taos/result.py
+5
-5
未找到文件。
src/connector/python/README.md
浏览文件 @
33f35846
...
...
@@ -41,6 +41,7 @@ cursor.execute("show databases")
results
=
cursor
.
fetchall
()
for
row
in
results
:
print
(
row
)
cursor
.
close
()
conn
.
close
()
```
...
...
@@ -57,8 +58,10 @@ result = conn.query("show databases")
num_of_fields
=
result
.
field_count
for
field
in
result
.
fields
:
print
(
field
)
for
row
in
result
:
print
(
row
)
result
.
close
()
conn
.
execute
(
"drop database pytest"
)
conn
.
close
()
...
...
@@ -81,6 +84,7 @@ def fetch_callback(p_param, p_result, num_of_rows):
p
.
contents
.
done
=
True
result
.
close
()
return
if
num_of_rows
<
0
:
p
.
contents
.
done
=
True
result
.
check_error
(
num_of_rows
)
...
...
@@ -90,6 +94,7 @@ def fetch_callback(p_param, p_result, num_of_rows):
for
row
in
result
.
rows_iter
(
num_of_rows
):
# print(row)
None
p
.
contents
.
count
+=
result
.
row_count
result
.
fetch_rows_a
(
fetch_callback
,
p_param
)
...
...
@@ -97,11 +102,13 @@ def fetch_callback(p_param, p_result, num_of_rows):
def
query_callback
(
p_param
,
p_result
,
code
):
# type: (c_void_p, c_void_p, c_int) -> None
if
p_result
==
None
:
if
p_result
is
None
:
return
result
=
TaosResult
(
p_result
)
if
code
==
0
:
result
.
fetch_rows_a
(
fetch_callback
,
p_param
)
result
.
check_error
(
code
)
...
...
@@ -120,6 +127,7 @@ def test_query(conn):
while
not
counter
.
done
:
print
(
"wait query callback"
)
time
.
sleep
(
1
)
print
(
counter
)
conn
.
close
()
...
...
@@ -182,6 +190,7 @@ result = conn.query("select * from log")
for
row
in
result
:
print
(
row
)
result
.
close
()
stmt
.
close
()
conn
.
close
()
...
...
@@ -237,18 +246,20 @@ result.close()
result
=
conn
.
query
(
"select * from log"
)
for
row
in
result
:
print
(
row
)
result
.
close
()
stmt
.
close
()
conn
.
close
()
```
### S
tatement API - Subscribe
### S
ubscription
```
python
import
taos
import
random
conn
=
taos
.
connect
()
dbname
=
"pytest_taos_subscribe
_callback
"
dbname
=
"pytest_taos_subscribe"
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
execute
(
"create database if not exists %s"
%
dbname
)
conn
.
select_db
(
dbname
)
...
...
@@ -256,7 +267,7 @@ conn.execute("create table if not exists log(ts timestamp, n int)")
for
i
in
range
(
10
):
conn
.
execute
(
"insert into log values(now, %d)"
%
i
)
sub
=
conn
.
subscribe
(
Tru
e
,
"test"
,
"select * from log"
,
1000
)
sub
=
conn
.
subscribe
(
Fals
e
,
"test"
,
"select * from log"
,
1000
)
print
(
"# consume from begin"
)
for
ts
,
n
in
sub
.
consume
():
print
(
ts
,
n
)
...
...
@@ -268,9 +279,18 @@ for i in range(5):
for
ts
,
n
in
result
:
print
(
ts
,
n
)
sub
.
close
(
True
)
print
(
"# keep progress consume"
)
sub
=
conn
.
subscribe
(
False
,
"test"
,
"select * from log"
,
1000
)
result
=
sub
.
consume
()
rows
=
result
.
fetch_all
()
# consume from latest subscription needs root privilege(for /var/lib/taos).
assert
result
.
row_count
==
0
print
(
"## consumed "
,
len
(
rows
),
"rows"
)
print
(
"# consume with a stop condition"
)
for
i
in
range
(
10
):
conn
.
execute
(
"insert into log values(now, %d)"
%
int
(
random
()
*
10
))
conn
.
execute
(
"insert into log values(now, %d)"
%
random
.
randint
(
0
,
10
))
result
=
sub
.
consume
()
try
:
ts
,
n
=
next
(
result
)
...
...
@@ -283,12 +303,13 @@ for i in range(10):
continue
sub
.
close
()
# sub.close()
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
#
conn.close()
```
### S
tatement API - Subscribe
asynchronously with callback
### S
ubscription
asynchronously with callback
```
python
from
taos
import
*
...
...
@@ -300,7 +321,7 @@ import time
def
subscribe_callback
(
p_sub
,
p_result
,
p_param
,
errno
):
# type: (c_void_p, c_void_p, c_void_p, c_int) -> None
print
(
"# fetch in callback"
)
result
=
TaosResult
(
p_result
)
result
=
TaosResult
(
c_void_p
(
p_result
)
)
result
.
check_error
(
errno
)
for
row
in
result
.
rows_iter
():
ts
,
n
=
row
()
...
...
@@ -311,42 +332,45 @@ def test_subscribe_callback(conn):
# type: (TaosConnection) -> None
dbname
=
"pytest_taos_subscribe_callback"
try
:
print
(
"drop if exists"
)
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
print
(
"create database"
)
conn
.
execute
(
"create database if not exists %s"
%
dbname
)
conn
.
select_db
(
dbname
)
conn
.
execute
(
"create table if not exists log(ts timestamp, n int)"
)
print
(
"create table"
)
# conn.execute("use %s" % dbname)
conn
.
execute
(
"create table if not exists %s.log(ts timestamp, n int)"
%
dbname
)
print
(
"# subscribe with callback"
)
sub
=
conn
.
subscribe
(
False
,
"test"
,
"select * from
log"
,
1000
,
subscribe_callback
)
sub
=
conn
.
subscribe
(
False
,
"test"
,
"select * from
%s.log"
%
dbname
,
1000
,
subscribe_callback
)
for
i
in
range
(
10
):
conn
.
execute
(
"insert into
log values(now, %d)"
%
i
)
conn
.
execute
(
"insert into
%s.log values(now, %d)"
%
(
dbname
,
i
)
)
time
.
sleep
(
0.7
)
sub
.
close
()
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
#
conn.close()
except
Exception
as
err
:
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
#
conn.close()
raise
err
if
__name__
==
"__main__"
:
test_subscribe_callback
(
connect
())
```
### St
atement API - St
ream
### Stream
```
python
from
taos
import
*
from
ctypes
import
*
import
time
def
stream_callback
(
p_param
,
p_result
,
p_row
):
# type: (c_void_p, c_void_p, c_void_p) -> None
if
p_result
==
None
or
p_row
==
None
:
if
p_result
is
None
or
p_row
is
None
:
return
result
=
TaosResult
(
p_result
)
row
=
TaosRow
(
result
,
p_row
)
...
...
@@ -355,13 +379,12 @@ def stream_callback(p_param, p_result, p_row):
p
=
cast
(
p_param
,
POINTER
(
Counter
))
p
.
contents
.
count
+=
count
print
(
"[%s] inserted %d in 5s, total count: %d"
%
(
ts
.
strftime
(
"%Y-%m-%d %H:%M:%S"
),
count
,
p
.
contents
.
count
))
except
Exception
as
err
:
print
(
err
)
raise
err
class
Counter
(
ctypes
.
Structure
):
class
Counter
(
Structure
):
_fields_
=
[
(
"count"
,
c_int
),
]
...
...
@@ -388,6 +411,7 @@ def test_stream(conn):
for
_
in
range
(
0
,
20
):
conn
.
execute
(
"insert into log values(now,0)(now+1s, 1)(now + 2s, 2)"
)
time
.
sleep
(
2
)
stream
.
close
()
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
...
...
@@ -399,12 +423,14 @@ def test_stream(conn):
if
__name__
==
"__main__"
:
test_stream
(
connect
())
```
### Insert with line protocol
```
python
import
taos
from
taos
import
SmlProtocol
,
SmlPrecision
conn
=
taos
.
connect
()
dbname
=
"pytest_line"
...
...
@@ -413,29 +439,22 @@ conn.execute("create database if not exists %s precision 'us'" % dbname)
conn
.
select_db
(
dbname
)
lines
=
[
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000ns'
,
'st,t1=4i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"pass it again",c2=true,c4=5f64,c5=5f64,c6=7u64 1626006933640000000'
,
'stf,t1=4i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"pass it again_stf",c2=false,c5=5f64,c6=7u64 1626006933641000000'
,
'st,t1=3i64,t2=4f64,t3="t3" c1=3i64,c3=L"pass",c2=false,c4=4f64 1626006833639000000'
,
]
conn
.
schemaless_insert
(
lines
,
0
,
"ns"
)
conn
.
schemaless_insert
(
lines
,
taos
.
SmlProtocol
.
LINE_PROTOCOL
,
taos
.
SmlPrecision
.
NOT_CONFIGURED
)
print
(
"inserted"
)
lines
=
[
'stf,t1=5i64,t3="t4",t2=5f64,t4=5f64 c1=3i64,c3=L"pass it again_stf",c2=false,c5=5f64,c6=7u64 1626006933641000000'
,
]
conn
.
schemaless_insert
(
lines
,
0
,
"ns"
)
conn
.
schemaless_insert
(
lines
,
taos
.
SmlProtocol
.
LINE_PROTOCOL
,
taos
.
SmlPrecision
.
NOT_CONFIGURED
)
result
=
conn
.
query
(
"show tables"
)
for
row
in
result
:
print
(
row
)
result
.
close
()
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
```
## License
- AGPL-3.0
## License
Keep same with
[
TDengine
](
https://github.com/taosdata/TDengine
)
.
We use MIT license for Python connector
.
src/connector/python/examples/query-async.py
浏览文件 @
33f35846
...
...
@@ -29,7 +29,7 @@ def fetch_callback(p_param, p_result, num_of_rows):
def
query_callback
(
p_param
,
p_result
,
code
):
# type: (c_void_p, c_void_p, c_int) -> None
if
p_result
==
None
:
if
p_result
is
None
:
return
result
=
TaosResult
(
p_result
)
if
code
==
0
:
...
...
src/connector/python/examples/stream.py
0 → 100644
浏览文件 @
33f35846
from
taos
import
*
from
ctypes
import
*
import
time
def
stream_callback
(
p_param
,
p_result
,
p_row
):
# type: (c_void_p, c_void_p, c_void_p) -> None
if
p_result
is
None
or
p_row
is
None
:
return
result
=
TaosResult
(
p_result
)
row
=
TaosRow
(
result
,
p_row
)
try
:
ts
,
count
=
row
()
p
=
cast
(
p_param
,
POINTER
(
Counter
))
p
.
contents
.
count
+=
count
print
(
"[%s] inserted %d in 5s, total count: %d"
%
(
ts
.
strftime
(
"%Y-%m-%d %H:%M:%S"
),
count
,
p
.
contents
.
count
))
except
Exception
as
err
:
print
(
err
)
raise
err
class
Counter
(
Structure
):
_fields_
=
[
(
"count"
,
c_int
),
]
def
__str__
(
self
):
return
"%d"
%
self
.
count
def
test_stream
(
conn
):
# type: (TaosConnection) -> None
dbname
=
"pytest_taos_stream"
try
:
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
execute
(
"create database if not exists %s"
%
dbname
)
conn
.
select_db
(
dbname
)
conn
.
execute
(
"create table if not exists log(ts timestamp, n int)"
)
result
=
conn
.
query
(
"select count(*) from log interval(5s)"
)
assert
result
.
field_count
==
2
counter
=
Counter
()
counter
.
count
=
0
stream
=
conn
.
stream
(
"select count(*) from log interval(5s)"
,
stream_callback
,
param
=
byref
(
counter
))
for
_
in
range
(
0
,
20
):
conn
.
execute
(
"insert into log values(now,0)(now+1s, 1)(now + 2s, 2)"
)
time
.
sleep
(
2
)
stream
.
close
()
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
except
Exception
as
err
:
conn
.
execute
(
"drop database if exists %s"
%
dbname
)
conn
.
close
()
raise
err
if
__name__
==
"__main__"
:
test_stream
(
connect
())
src/connector/python/taos/__init__.py
浏览文件 @
33f35846
...
...
@@ -86,7 +86,7 @@ def fetch_callback(p_param, p_result, num_of_rows):
def query_callback(p_param, p_result, code):
# type: (c_void_p, c_void_p, c_int) -> None
if p_result
==
None:
if p_result
is
None:
return
result = TaosResult(p_result)
if code == 0:
...
...
@@ -335,7 +335,7 @@ from ctypes import *
def stream_callback(p_param, p_result, p_row):
# type: (c_void_p, c_void_p, c_void_p) -> None
if p_result
== None or p_row ==
None:
if p_result
is None or p_row is
None:
return
result = TaosResult(p_result)
row = TaosRow(result, p_row)
...
...
src/connector/python/taos/bind.py
浏览文件 @
33f35846
...
...
@@ -317,7 +317,7 @@ class TaosMultiBind(ctypes.Structure):
def
_str_to_buffer
(
self
,
values
):
self
.
num
=
len
(
values
)
is_null
=
[
1
if
v
==
None
else
0
for
v
in
values
]
is_null
=
[
1
if
v
is
None
else
0
for
v
in
values
]
self
.
is_null
=
cast
((
c_byte
*
self
.
num
)(
*
is_null
),
c_char_p
)
if
sum
(
is_null
)
==
self
.
num
:
...
...
src/connector/python/taos/cinterface.py
浏览文件 @
33f35846
...
...
@@ -373,9 +373,9 @@ def taos_fetch_block(result, fields=None, field_count=None):
if
num_of_rows
==
0
:
return
None
,
0
precision
=
taos_result_precision
(
result
)
if
fields
==
None
:
if
fields
is
None
:
fields
=
taos_fetch_fields
(
result
)
if
field_count
==
None
:
if
field_count
is
None
:
field_count
=
taos_field_count
(
result
)
blocks
=
[
None
]
*
field_count
fieldLen
=
taos_fetch_lengths
(
result
,
field_count
)
...
...
@@ -466,7 +466,7 @@ def taos_fetch_lengths(result, field_count=None):
# type: (c_void_p, int) -> Array[int]
"""Make sure to call taos_fetch_row or taos_fetch_block before fetch_lengths"""
lens
=
_libtaos
.
taos_fetch_lengths
(
result
)
if
field_count
==
None
:
if
field_count
is
None
:
field_count
=
taos_field_count
(
result
)
if
not
lens
:
raise
OperationalError
(
"field length empty, use taos_fetch_row/block before it"
)
...
...
@@ -823,7 +823,7 @@ def taos_stmt_use_result(stmt):
@stmt: TAOS_STMT*
"""
result
=
c_void_p
(
_libtaos
.
taos_stmt_use_result
(
stmt
))
if
result
==
None
:
if
result
is
None
:
raise
StatementError
(
taos_stmt_errstr
(
stmt
))
return
result
...
...
src/connector/python/taos/result.py
浏览文件 @
33f35846
...
...
@@ -41,7 +41,7 @@ class TaosResult(object):
if
self
.
_result
is
None
or
self
.
fields
is
None
:
raise
OperationalError
(
"Invalid use of fetch iterator"
)
if
self
.
_block
==
None
or
self
.
_block_iter
>=
self
.
_block_length
:
if
self
.
_block
is
None
or
self
.
_block_iter
>=
self
.
_block_length
:
self
.
_block
,
self
.
_block_length
=
self
.
fetch_block
()
self
.
_block_iter
=
0
# self._row_count += self._block_length
...
...
@@ -55,7 +55,7 @@ class TaosResult(object):
"""fields definitions of the current result"""
if
self
.
_result
is
None
:
raise
ResultError
(
"no result object setted"
)
if
self
.
_fields
==
None
:
if
self
.
_fields
is
None
:
self
.
_fields
=
taos_fetch_fields
(
self
.
_result
)
return
self
.
_fields
...
...
@@ -72,7 +72,7 @@ class TaosResult(object):
@
property
def
precision
(
self
):
if
self
.
_precision
==
None
:
if
self
.
_precision
is
None
:
self
.
_precision
=
taos_result_precision
(
self
.
_result
)
return
self
.
_precision
...
...
@@ -114,7 +114,7 @@ class TaosResult(object):
if
self
.
_result
is
None
:
raise
OperationalError
(
"Invalid use of fetchall"
)
if
self
.
_fields
==
None
:
if
self
.
_fields
is
None
:
self
.
_fields
=
taos_fetch_fields
(
self
.
_result
)
buffer
=
[[]
for
i
in
range
(
len
(
self
.
_fields
))]
self
.
_row_count
=
0
...
...
@@ -150,7 +150,7 @@ class TaosResult(object):
return
taos_errstr
(
self
.
_result
)
def
check_error
(
self
,
errno
=
None
,
close
=
True
):
if
errno
==
None
:
if
errno
is
None
:
errno
=
self
.
errno
()
if
errno
!=
0
:
msg
=
self
.
errstr
()
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录