linter

Author: rekby

tests/topics/test_topic_reader.py

@@ -168,8 +168,9 @@ def decode(b: bytes):
 @pytest.mark.asyncio
 class TestBugFixesAsync:
     @pytest.mark.skip("LOGBROKER-8319")
-    async def test_issue_297_bad_handle_stop_partition(self, driver, topic_consumer, topic_with_two_partitions_path: str):
-
+    async def test_issue_297_bad_handle_stop_partition(
+        self, driver, topic_consumer, topic_with_two_partitions_path: str
+    ):
         async def wait(fut):
             return await asyncio.wait_for(fut, timeout=10)
 
@@ -187,10 +188,13 @@ async def wait(fut):
         await wait(reader0.receive_message())
 
         # Start second reader for same topic, same consumer, partition 1
-        reader1 = driver.topic_client.reader(ydb.TopicReaderSelector(
-            path=topic,
-            partitions=1,
-        ), consumer=topic_consumer)
+        reader1 = driver.topic_client.reader(
+            ydb.TopicReaderSelector(
+                path=topic,
+                partitions=1,
+            ),
+            consumer=topic_consumer,
+        )
 
         await asyncio.sleep(0.1)
 

ydb/_grpc/grpcwrapper/ydb_topic.py

@@ -601,15 +601,16 @@ class PartitionSessionStatusResponse(IFromProto):
         write_time_high_watermark: float
 
         @staticmethod
-        def from_proto(msg: ydb_topic_pb2.StreamReadMessage.PartitionSessionStatusResponse) -> "StreamReadMessage.PartitionSessionStatusResponse":
+        def from_proto(
+            msg: ydb_topic_pb2.StreamReadMessage.PartitionSessionStatusResponse,
+        ) -> "StreamReadMessage.PartitionSessionStatusResponse":
             return StreamReadMessage.PartitionSessionStatusResponse(
                 partition_session_id=msg.partition_session_id,
                 partition_offsets=OffsetsRange.from_proto(msg.partition_offsets),
                 committed_offset=msg.committed_offset,
                 write_time_high_watermark=msg.write_time_high_watermark,
             )
 
-
     @dataclass
     class StartPartitionSessionRequest(IFromProto):
         partition_session: "StreamReadMessage.PartitionSession"
@@ -650,14 +651,15 @@ class StopPartitionSessionRequest(IFromProto):
         committed_offset: int
 
         @staticmethod
-        def from_proto(msg: ydb_topic_pb2.StreamReadMessage.StopPartitionSessionRequest) -> StreamReadMessage.StopPartitionSessionRequest:
+        def from_proto(
+            msg: ydb_topic_pb2.StreamReadMessage.StopPartitionSessionRequest,
+        ) -> StreamReadMessage.StopPartitionSessionRequest:
             return StreamReadMessage.StopPartitionSessionRequest(
                 partition_session_id=msg.partition_session_id,
                 graceful=msg.graceful,
                 committed_offset=msg.committed_offset,
             )
 
-
     @dataclass
     class StopPartitionSessionResponse:
         partition_session_id: int
@@ -723,7 +725,7 @@ def from_proto(
                     server_status=server_status,
                     server_message=StreamReadMessage.StopPartitionSessionRequest.from_proto(
                         msg.stop_partition_session_request
-                    )
+                    ),
                 )
             elif mess_type == "update_token_response":
                 return StreamReadMessage.FromServer(
@@ -735,7 +737,7 @@ def from_proto(
                     server_status=server_status,
                     server_message=StreamReadMessage.PartitionSessionStatusResponse.from_proto(
                         msg.partition_session_status_response
-                    )
+                    ),
                 )
 
 

ydb/_topic_reader/topic_reader_asyncio.py

@@ -31,6 +31,7 @@
 
 _module_logger = logging.getLogger(__name__)
 
+
 class TopicReaderError(YdbError):
     pass
 

ydb/_topic_reader/topic_reader_asyncio_test.py

@@ -1140,13 +1140,15 @@ def set_logged(*args, **kwargs):
         stream_reader._logger.exception = mock.Mock(side_effect=set_logged)
 
         # noinspection PyTypeChecker
-        stream.from_server.put_nowait(StreamReadMessage.FromServer(
-            server_status=ServerStatus(
-                status=issues.StatusCode.SUCCESS,
-                issues=[],
-            ),
-            server_message=TestMessage(),
-        ))
+        stream.from_server.put_nowait(
+            StreamReadMessage.FromServer(
+                server_status=ServerStatus(
+                    status=issues.StatusCode.SUCCESS,
+                    issues=[],
+                ),
+                server_message=TestMessage(),
+            )
+        )
         await wait_for_fast(logged.wait())
 
         stream_reader._logger.exception.assert_called_once()