Merge remote-tracking branch 'upstream/main' into main

Author: RaileyHartheim

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added comments about calls encoders/decoders in parallel from multiply threads
+
 ## 3.5.1 ##
 * Fixed access to connection if connection cannot be found by node id
 

setup.py

@@ -13,7 +13,7 @@
 
 setuptools.setup(
     name="ydb",
-    version="3.5.0",  # AUTOVERSION
+    version="3.5.1",  # AUTOVERSION
     description="YDB Python SDK",
     author="Yandex LLC",
     author_email="ydb@yandex-team.ru",

ydb/aio/pool.py

@@ -1,11 +1,12 @@
 import asyncio
 import logging
 import random
+import typing
 
 from ydb import issues
 from ydb.pool import ConnectionsCache as _ConnectionsCache, IConnectionPool
 
-from .connection import Connection
+from .connection import Connection, EndpointKey
 
 from . import resolver
 
@@ -21,7 +22,7 @@ def __init__(self, use_all_nodes: bool = False):
 
         self._fast_fail_error = None
 
-    async def get(self, preferred_endpoint=None, fast_fail=False, wait_timeout=10):
+    async def get(self, preferred_endpoint: typing.Optional[EndpointKey] = None, fast_fail=False, wait_timeout=10):
 
         if fast_fail:
             await asyncio.wait_for(self._fast_fail_event.wait(), timeout=wait_timeout)

ydb/pool.py

@@ -5,11 +5,12 @@
 from concurrent import futures
 import collections
 import random
+import typing
 
 from . import connection as connection_impl, issues, resolver, _utilities, tracing
 from abc import abstractmethod
 
-from .connection import Connection
+from .connection import Connection, EndpointKey
 
 logger = logging.getLogger(__name__)
 
@@ -123,7 +124,7 @@ def subscribe(self):
             return subscription
 
     @tracing.with_trace()
-    def get(self, preferred_endpoint=None) -> Connection:
+    def get(self, preferred_endpoint: typing.Optional[EndpointKey] = None) -> Connection:
         with self.lock:
             if preferred_endpoint is not None and preferred_endpoint.node_id in self.connections_by_node_id:
                 return self.connections_by_node_id[preferred_endpoint.node_id]

ydb/topic.py

@@ -170,8 +170,11 @@ def reader(
         consumer: str,
         buffer_size_bytes: int = 50 * 1024 * 1024,
         # decoders: map[codec_code] func(encoded_bytes)->decoded_bytes
+        # the func will be called from multiply threads in parallel
         decoders: Union[Mapping[int, Callable[[bytes], bytes]], None] = None,
-        decoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
+        # custom decoder executor for call builtin and custom decoders. If None - use shared executor pool.
+        # if max_worker in the executor is 1 - then decoders will be called from the thread without parallel
+        decoder_executor: Optional[concurrent.futures.Executor] = None,
     ) -> TopicReaderAsyncIO:
 
         if not decoder_executor:
@@ -194,8 +197,12 @@ def writer(
         auto_seqno: bool = True,
         auto_created_at: bool = True,
         codec: Optional[TopicCodec] = None,  # default mean auto-select
+        # encoders: map[codec_code] func(encoded_bytes)->decoded_bytes
+        # the func will be called from multiply threads in parallel.
         encoders: Optional[Mapping[_ydb_topic_public_types.PublicCodec, Callable[[bytes], bytes]]] = None,
-        encoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
+        # custom encoder executor for call builtin and custom decoders. If None - use shared executor pool.
+        # If max_worker in the executor is 1 - then encoders will be called from the thread without parallel.
+        encoder_executor: Optional[concurrent.futures.Executor] = None,
     ) -> TopicWriterAsyncIO:
         args = locals().copy()
         del args["self"]
@@ -319,7 +326,10 @@ def reader(
         consumer: str,
         buffer_size_bytes: int = 50 * 1024 * 1024,
         # decoders: map[codec_code] func(encoded_bytes)->decoded_bytes
+        # the func will be called from multiply threads in parallel
         decoders: Union[Mapping[int, Callable[[bytes], bytes]], None] = None,
+        # custom decoder executor for call builtin and custom decoders. If None - use shared executor pool.
+        # if max_worker in the executor is 1 - then decoders will be called from the thread without parallel
         decoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
     ) -> TopicReader:
         if not decoder_executor:
@@ -343,7 +353,11 @@ def writer(
         auto_seqno: bool = True,
         auto_created_at: bool = True,
         codec: Optional[TopicCodec] = None,  # default mean auto-select
+        # encoders: map[codec_code] func(encoded_bytes)->decoded_bytes
+        # the func will be called from multiply threads in parallel.
         encoders: Optional[Mapping[_ydb_topic_public_types.PublicCodec, Callable[[bytes], bytes]]] = None,
+        # custom encoder executor for call builtin and custom decoders. If None - use shared executor pool.
+        # If max_worker in the executor is 1 - then encoders will be called from the thread without parallel.
         encoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
     ) -> TopicWriter:
         args = locals().copy()

ydb/ydb_version.py

@@ -1 +1 @@
-VERSION = "3.5.0"
+VERSION = "3.5.1"