This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch refactor_python_package
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/refactor_python_package by
this push:
new f05a7f140a5 fix code style
f05a7f140a5 is described below
commit f05a7f140a56b5da468e927570ba2f2e355a00c7
Author: HTHou <[email protected]>
AuthorDate: Thu Aug 31 18:13:58 2023 +0800
fix code style
---
iotdb-client/client-py/SessionPoolExample.py | 21 ++++----
iotdb-client/client-py/iotdb/IoTDBContainer.py | 9 ++--
iotdb-client/client-py/iotdb/SessionPool.py | 60 +++++++++++++++++------
iotdb-client/client-py/tests/test_session.py | 60 +++++++++++++----------
iotdb-client/client-py/tests/test_session_pool.py | 33 ++++++++++---
5 files changed, 121 insertions(+), 62 deletions(-)
diff --git a/iotdb-client/client-py/SessionPoolExample.py
b/iotdb-client/client-py/SessionPoolExample.py
index 29bf8b160d6..9230634ba6f 100644
--- a/iotdb-client/client-py/SessionPoolExample.py
+++ b/iotdb-client/client-py/SessionPoolExample.py
@@ -45,13 +45,13 @@ def prepare_data():
"root.test.d1.s0",
"root.test.d1.s1",
"root.test.d1.s2",
- "root.test.d1.s3"
+ "root.test.d1.s3",
]
data_type_lst = [
TSDataType.BOOLEAN,
TSDataType.INT32,
TSDataType.INT64,
- TSDataType.FLOAT
+ TSDataType.FLOAT,
]
encoding_lst = [TSEncoding.PLAIN for _ in range(len(data_type_lst))]
compressor_lst = [Compressor.SNAPPY for _ in range(len(data_type_lst))]
@@ -76,11 +76,7 @@ def insert_data(num: int):
device = ["root.test.d" + str(num)]
measurements = ["s0", "s1", "s2"]
values = [False, 10, 11]
- data_types = [
- TSDataType.BOOLEAN,
- TSDataType.INT32,
- TSDataType.INT64
- ]
+ data_types = [TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64]
# Get a session from the pool
session = session_pool.get_session()
session.insert_records(device, [1], [measurements], [data_types], [values])
@@ -119,9 +115,14 @@ ip = "127.0.0.1"
port = "6667"
username = "root"
password = "root"
-pool_config = PoolConfig(node_urls=["127.0.0.1:6667", "127.0.0.1:6668",
"127.0.0.1:6669"], user_name=username,
- password=password, fetch_size=1024,
- time_zone="UTC+8", max_retry=3)
+pool_config = PoolConfig(
+ node_urls=["127.0.0.1:6667", "127.0.0.1:6668", "127.0.0.1:6669"],
+ user_name=username,
+ password=password,
+ fetch_size=1024,
+ time_zone="UTC+8",
+ max_retry=3,
+)
max_pool_size = 5
wait_timeout_in_ms = 3000
diff --git a/iotdb-client/client-py/iotdb/IoTDBContainer.py
b/iotdb-client/client-py/iotdb/IoTDBContainer.py
index 2f52f17a930..51ac52c05ed 100644
--- a/iotdb-client/client-py/iotdb/IoTDBContainer.py
+++ b/iotdb-client/client-py/iotdb/IoTDBContainer.py
@@ -42,11 +42,12 @@ class IoTDBContainer(DockerContainer):
self.get_container_host_ip(), self.get_exposed_port(6667), "root",
"root"
)
session.open(False)
- with session.execute_statement(
- "SHOW CLUSTER"
- ) as session_data_set:
+ with session.execute_statement("SHOW CLUSTER") as session_data_set:
while session_data_set.has_next():
- if session_data_set.next().get_fields()[2].get_string_value()
!= "Running":
+ if (
+ session_data_set.next().get_fields()[2].get_string_value()
+ != "Running"
+ ):
raise ContainerStartException("IoTDB is not started")
session.close()
diff --git a/iotdb-client/client-py/iotdb/SessionPool.py
b/iotdb-client/client-py/iotdb/SessionPool.py
index 4a7bdc3ae39..6f1d7580796 100644
--- a/iotdb-client/client-py/iotdb/SessionPool.py
+++ b/iotdb-client/client-py/iotdb/SessionPool.py
@@ -31,15 +31,25 @@ logger = logging.getLogger("IoTDB")
class PoolConfig(object):
- def __init__(self, host: str = None, port: str = None, user_name: str =
None, password: str = None,
- node_urls: list = None,
- fetch_size: int = DEFAULT_FETCH_SIZE, time_zone: str =
DEFAULT_TIME_ZONE,
- max_retry: int = DEFAULT_MAX_RETRY, enable_compression: bool
= False):
+ def __init__(
+ self,
+ host: str = None,
+ port: str = None,
+ user_name: str = None,
+ password: str = None,
+ node_urls: list = None,
+ fetch_size: int = DEFAULT_FETCH_SIZE,
+ time_zone: str = DEFAULT_TIME_ZONE,
+ max_retry: int = DEFAULT_MAX_RETRY,
+ enable_compression: bool = False,
+ ):
self.host = host
self.port = port
if node_urls is None:
if host is None or port is None:
- raise ValueError("(host,port) and node_urls cannot be None at
the same time.")
+ raise ValueError(
+ "(host,port) and node_urls cannot be None at the same
time."
+ )
node_urls = []
self.node_urls = node_urls
self.user_name = user_name
@@ -51,8 +61,9 @@ class PoolConfig(object):
class SessionPool(object):
-
- def __init__(self, pool_config: PoolConfig, max_pool_size: int,
wait_timeout_in_ms: int):
+ def __init__(
+ self, pool_config: PoolConfig, max_pool_size: int, wait_timeout_in_ms:
int
+ ):
self.__pool_config = pool_config
self.__max_pool_size = max_pool_size
self.__wait_timeout_in_ms = wait_timeout_in_ms / 1000
@@ -63,13 +74,23 @@ class SessionPool(object):
def __construct_session(self) -> Session:
if len(self.__pool_config.node_urls) > 0:
- session =
Session.init_from_node_urls(self.__pool_config.node_urls,
self.__pool_config.user_name,
- self.__pool_config.password,
self.__pool_config.fetch_size,
- self.__pool_config.time_zone)
+ session = Session.init_from_node_urls(
+ self.__pool_config.node_urls,
+ self.__pool_config.user_name,
+ self.__pool_config.password,
+ self.__pool_config.fetch_size,
+ self.__pool_config.time_zone,
+ )
else:
- session = Session(self.__pool_config.host,
self.__pool_config.port, self.__pool_config.user_name,
- self.__pool_config.password,
self.__pool_config.fetch_size, self.__pool_config.time_zone)
+ session = Session(
+ self.__pool_config.host,
+ self.__pool_config.port,
+ self.__pool_config.user_name,
+ self.__pool_config.password,
+ self.__pool_config.fetch_size,
+ self.__pool_config.time_zone,
+ )
session.open(self.__pool_config.enable_compression)
return session
@@ -97,8 +118,11 @@ class SessionPool(object):
break
else:
if time.time() - start > self.__wait_timeout_in_ms:
- raise TimeoutError("Wait to get session timeout in
SessionPool, current pool size: {0}"
- .format(self.__max_pool_size))
+ raise TimeoutError(
+ "Wait to get session timeout in SessionPool,
current pool size: {0}".format(
+ self.__max_pool_size
+ )
+ )
time.sleep(1)
session = self.__poll_session()
@@ -115,7 +139,9 @@ class SessionPool(object):
def put_back(self, session: Session):
if self.__closed:
- raise ConnectionError("SessionPool has already been closed, please
close the session manually.")
+ raise ConnectionError(
+ "SessionPool has already been closed, please close the session
manually."
+ )
if session.is_open():
self.__queue.put(session)
@@ -134,7 +160,9 @@ class SessionPool(object):
logger.info("SessionPool has been closed successfully.")
-def create_session_pool(pool_config: PoolConfig, max_pool_size: int,
wait_timeout_in_ms: int) -> SessionPool:
+def create_session_pool(
+ pool_config: PoolConfig, max_pool_size: int, wait_timeout_in_ms: int
+) -> SessionPool:
if max_pool_size <= 0:
max_pool_size = multiprocessing.cpu_count() * DEFAULT_MULTIPIE
return SessionPool(pool_config, max_pool_size, wait_timeout_in_ms)
diff --git a/iotdb-client/client-py/tests/test_session.py
b/iotdb-client/client-py/tests/test_session.py
index 5118fe90658..197b9b9de52 100644
--- a/iotdb-client/client-py/tests/test_session.py
+++ b/iotdb-client/client-py/tests/test_session.py
@@ -59,8 +59,16 @@ def session_test(use_session_pool=False):
db: IoTDBContainer
if use_session_pool:
- pool_config = PoolConfig(db.get_container_host_ip(),
db.get_exposed_port(6667), "root", "root", None, 1024,
- "Asia/Shanghai", 3)
+ pool_config = PoolConfig(
+ db.get_container_host_ip(),
+ db.get_exposed_port(6667),
+ "root",
+ "root",
+ None,
+ 1024,
+ "Asia/Shanghai",
+ 3,
+ )
session_pool = create_session_pool(pool_config, 1, 3000)
session = session_pool.get_session()
else:
@@ -170,14 +178,14 @@ def session_test(use_session_pool=False):
# delete time series
if (
- session.delete_time_series(
- [
- "root.sg_test_01.d_01.s_07",
- "root.sg_test_01.d_01.s_08",
- "root.sg_test_01.d_01.s_09",
- ]
- )
- < 0
+ session.delete_time_series(
+ [
+ "root.sg_test_01.d_01.s_07",
+ "root.sg_test_01.d_01.s_08",
+ "root.sg_test_01.d_01.s_09",
+ ]
+ )
+ < 0
):
test_fail()
print_message("delete time series failed")
@@ -213,10 +221,10 @@ def session_test(use_session_pool=False):
TSDataType.TEXT,
]
if (
- session.insert_record(
- "root.sg_test_01.d_01", 1, measurements_, data_types_,
values_
- )
- < 0
+ session.insert_record(
+ "root.sg_test_01.d_01", 1, measurements_, data_types_, values_
+ )
+ < 0
):
test_fail()
print_message("insert record failed")
@@ -233,10 +241,10 @@ def session_test(use_session_pool=False):
data_type_list_ = [data_types_, data_types_]
device_ids_ = ["root.sg_test_01.d_01", "root.sg_test_01.d_02"]
if (
- session.insert_records(
- device_ids_, [2, 3], measurements_list_, data_type_list_,
values_list_
- )
- < 0
+ session.insert_records(
+ device_ids_, [2, 3], measurements_list_, data_type_list_,
values_list_
+ )
+ < 0
):
test_fail()
print_message("insert records failed")
@@ -353,14 +361,14 @@ def session_test(use_session_pool=False):
values_list = [[False, 22, 33], [True, 1, 23], [False, 15, 26]]
if (
- session.insert_records_of_one_device(
- "root.sg_test_01.d_01",
- time_list,
- measurements_list,
- data_types_list,
- values_list,
- )
- < 0
+ session.insert_records_of_one_device(
+ "root.sg_test_01.d_01",
+ time_list,
+ measurements_list,
+ data_types_list,
+ values_list,
+ )
+ < 0
):
test_fail()
print_message("insert records of one device failed")
diff --git a/iotdb-client/client-py/tests/test_session_pool.py
b/iotdb-client/client-py/tests/test_session_pool.py
index fc71882fadc..43351be2516 100644
--- a/iotdb-client/client-py/tests/test_session_pool.py
+++ b/iotdb-client/client-py/tests/test_session_pool.py
@@ -27,8 +27,16 @@ def test_session_pool():
with IoTDBContainer(CONTAINER_NAME) as db:
db: IoTDBContainer
max_pool_size = 2
- pool_config = PoolConfig(db.get_container_host_ip(),
db.get_exposed_port(6667), "root", "root",
- [], 1024, "Asia/Shanghai", 3)
+ pool_config = PoolConfig(
+ db.get_container_host_ip(),
+ db.get_exposed_port(6667),
+ "root",
+ "root",
+ [],
+ 1024,
+ "Asia/Shanghai",
+ 3,
+ )
session_pool = create_session_pool(pool_config, max_pool_size, 3000)
session = session_pool.get_session()
assert session.is_open() is True
@@ -39,7 +47,11 @@ def test_session_pool():
session_pool.get_session()
except TimeoutError as e:
timeout = True
- assert str(e) == "Wait to get session timeout in SessionPool,
current pool size: " + str(max_pool_size)
+ assert str(
+ e
+ ) == "Wait to get session timeout in SessionPool, current pool
size: " + str(
+ max_pool_size
+ )
assert timeout is True
Thread(target=lambda: session_pool.put_back(session2)).start()
@@ -60,7 +72,10 @@ def test_session_pool():
session_pool.put_back(session3)
except ConnectionError as e:
is_closed = True
- assert str(e) == "SessionPool has already been closed, please
close the session manually."
+ assert (
+ str(e)
+ == "SessionPool has already been closed, please close the
session manually."
+ )
assert is_closed is True
@@ -69,7 +84,9 @@ def test_session_pool_by_node_urls():
db: IoTDBContainer
node_url = db.get_container_host_ip() + ":" +
str(db.get_exposed_port(6667))
max_pool_size = 1
- pool_config = PoolConfig(node_urls=[node_url], user_name="root",
password="root")
+ pool_config = PoolConfig(
+ node_urls=[node_url], user_name="root", password="root"
+ )
session_pool = create_session_pool(pool_config, max_pool_size, 3000)
session = session_pool.get_session()
@@ -79,7 +96,11 @@ def test_session_pool_by_node_urls():
session_pool.get_session()
except TimeoutError as e:
timeout = True
- assert str(e) == "Wait to get session timeout in SessionPool,
current pool size: " + str(max_pool_size)
+ assert str(
+ e
+ ) == "Wait to get session timeout in SessionPool, current pool
size: " + str(
+ max_pool_size
+ )
assert timeout is True
