linter

Author: rekby

ydb/_topic_common/common.py

@@ -68,13 +68,17 @@ class CallFromSyncToAsync:
     def __init__(self, loop: asyncio.AbstractEventLoop):
         self._loop = loop
 
-    def unsafe_call_with_future(self, coro: typing.Coroutine) -> concurrent.futures.Future:
+    def unsafe_call_with_future(
+        self, coro: typing.Coroutine
+    ) -> concurrent.futures.Future:
         """
         returned result from coro may be lost
         """
         return asyncio.run_coroutine_threadsafe(coro, self._loop)
 
-    def unsafe_call_with_result(self, coro: typing.Coroutine, timeout: typing.Union[int, float, None]):
+    def unsafe_call_with_result(
+        self, coro: typing.Coroutine, timeout: typing.Union[int, float, None]
+    ):
         """
         returned result from coro may be lost by race future cancel by timeout and return value from coroutine
         """
@@ -99,7 +103,7 @@ async def call_coro():
             try:
                 res = await asyncio.wait_for(task, timeout)
                 return res
-            except BaseException as err:
+            except asyncio.TimeoutError:
                 try:
                     res = await task
                     return res
@@ -109,7 +113,6 @@ async def call_coro():
                 # return builtin TimeoutError instead of asyncio.TimeoutError
                 raise TimeoutError()
 
-
         return asyncio.run_coroutine_threadsafe(call_coro(), self._loop).result()
 
     def _safe_call_fast(self, coro: typing.Coroutine):

ydb/_topic_reader/topic_reader_sync.py

@@ -1,7 +1,7 @@
 import asyncio
 import concurrent.futures
 import typing
-from typing import List, Union, Iterable, Optional, Coroutine
+from typing import List, Union, Iterable, Optional
 
 from ydb._grpc.grpcwrapper.common_utils import SupportedDriverType
 from ydb._topic_common.common import _get_shared_event_loop, CallFromSyncToAsync
@@ -24,11 +24,11 @@ class TopicReaderSync:
     _closed: bool
 
     def __init__(
-            self,
-            driver: SupportedDriverType,
-            settings: PublicReaderSettings,
-            *,
-            eventloop: Optional[asyncio.AbstractEventLoop] = None,
+        self,
+        driver: SupportedDriverType,
+        settings: PublicReaderSettings,
+        *,
+        eventloop: Optional[asyncio.AbstractEventLoop] = None,
     ):
         self._closed = False
 
@@ -70,7 +70,7 @@ async def sessions_stat(self) -> List[SessionStat]:
         raise NotImplementedError()
 
     def messages(
-            self, *, timeout: Union[float, None] = None
+        self, *, timeout: Union[float, None] = None
     ) -> Iterable[PublicMessage]:
         """
         todo?
@@ -104,11 +104,11 @@ def async_wait_message(self) -> concurrent.futures.Future:
         raise NotImplementedError()
 
     def batches(
-            self,
-            *,
-            max_messages: Union[int, None] = None,
-            max_bytes: Union[int, None] = None,
-            timeout: Union[float, None] = None,
+        self,
+        *,
+        max_messages: Union[int, None] = None,
+        max_bytes: Union[int, None] = None,
+        timeout: Union[float, None] = None,
     ) -> Iterable[PublicBatch]:
         """
         Block until receive new batch.
@@ -120,11 +120,11 @@ def batches(
         raise NotImplementedError()
 
     def receive_batch(
-            self,
-            *,
-            max_messages: typing.Union[int, None] = None,
-            max_bytes: typing.Union[int, None] = None,
-            timeout: Union[float, None] = None,
+        self,
+        *,
+        max_messages: typing.Union[int, None] = None,
+        max_bytes: typing.Union[int, None] = None,
+        timeout: Union[float, None] = None,
     ) -> Union[PublicBatch, None]:
         """
         Get one messages batch from reader
@@ -136,10 +136,15 @@ def receive_batch(
         self._check_closed()
 
         return self._caller.safe_call_with_result(
-            self._async_reader.receive_batch(max_messages=max_messages, max_bytes=max_bytes),
-            timeout)
-
-    def commit(self, mess: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]):
+            self._async_reader.receive_batch(
+                max_messages=max_messages, max_bytes=max_bytes
+            ),
+            timeout,
+        )
+
+    def commit(
+        self, mess: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]
+    ):
         """
         Put commit message to internal buffer.
 
@@ -151,7 +156,7 @@ def commit(self, mess: typing.Union[datatypes.PublicMessage, datatypes.PublicBat
         self._caller.call_sync(self._async_reader.commit(mess))
 
     def commit_with_ack(
-            self, mess: ICommittable, timeout: typing.Union[int, float, None] = None
+        self, mess: ICommittable, timeout: typing.Union[int, float, None] = None
     ) -> Union[CommitResult, List[CommitResult]]:
         """
         write commit message to a buffer and wait ack from the server.
@@ -160,15 +165,21 @@ def commit_with_ack(
         """
         self._check_closed()
 
-        return self._caller.unsafe_call_with_result(self._async_reader.commit_with_ack(mess), timeout)
+        return self._caller.unsafe_call_with_result(
+            self._async_reader.commit_with_ack(mess), timeout
+        )
 
-    def async_commit_with_ack(self, mess: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]) -> concurrent.futures.Future:
+    def async_commit_with_ack(
+        self, mess: typing.Union[datatypes.PublicMessage, datatypes.PublicBatch]
+    ) -> concurrent.futures.Future:
         """
         write commit message to a buffer and return Future for wait result.
         """
         self._check_closed()
 
-        return self._caller.unsafe_call_with_future(self._async_reader.commit_with_ack(mess))
+        return self._caller.unsafe_call_with_future(
+            self._async_reader.commit_with_ack(mess)
+        )
 
     def async_flush(self) -> concurrent.futures.Future:
         """

ydb/_topic_writer/topic_writer_sync.py

@@ -3,19 +3,23 @@
 import asyncio
 import typing
 from concurrent.futures import Future
-from typing import Union, List, Optional, Coroutine
+from typing import Union, List, Optional
 
 from .._grpc.grpcwrapper.common_utils import SupportedDriverType
 from .topic_writer import (
     PublicWriterSettings,
-    TopicWriterError,
     PublicWriterInitInfo,
     PublicWriteResult,
-    Message, TopicWriterClosedError,
+    Message,
+    TopicWriterClosedError,
 )
 
 from .topic_writer_asyncio import WriterAsyncIO
-from .._topic_common.common import _get_shared_event_loop, TimeoutType, CallFromSyncToAsync
+from .._topic_common.common import (
+    _get_shared_event_loop,
+    TimeoutType,
+    CallFromSyncToAsync,
+)
 
 
 class WriterSync:
@@ -43,21 +47,27 @@ def __init__(
         async def create_async_writer():
             return WriterAsyncIO(driver, settings)
 
-        self._async_writer = self._caller.safe_call_with_result(create_async_writer(), None)
+        self._async_writer = self._caller.safe_call_with_result(
+            create_async_writer(), None
+        )
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
-    def close(self, *, flush: bool = True, timeout: typing.Union[int, float, None] = None):
+    def close(
+        self, *, flush: bool = True, timeout: typing.Union[int, float, None] = None
+    ):
         if self._closed:
             return
 
         self._closed = True
 
-        self._caller.safe_call_with_result(self._async_writer.close(flush=flush), timeout)
+        self._caller.safe_call_with_result(
+            self._async_writer.close(flush=flush), timeout
+        )
 
     def _check_closed(self):
         if self._closed:
@@ -81,7 +91,9 @@ def async_wait_init(self) -> Future[PublicWriterInitInfo]:
     def wait_init(self, *, timeout: TimeoutType = None) -> PublicWriterInitInfo:
         self._check_closed()
 
-        return self._caller.unsafe_call_with_result(self._async_writer.wait_init(), timeout)
+        return self._caller.unsafe_call_with_result(
+            self._async_writer.wait_init(), timeout
+        )
 
     def write(
         self,
@@ -98,7 +110,9 @@ def async_write_with_ack(
     ) -> Future[Union[PublicWriteResult, List[PublicWriteResult]]]:
         self._check_closed()
 
-        return self._caller.unsafe_call_with_future(self._async_writer.write_with_ack(messages))
+        return self._caller.unsafe_call_with_future(
+            self._async_writer.write_with_ack(messages)
+        )
 
     def write_with_ack(
         self,