fix linter

Author: rekby

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Initial implementation of topic reader
+
 ## 3.0.1b3 ##
 * Fix error of check retriable error for idempotent operations (error exist since 2.12.1)
 

tests/conftest.py

@@ -135,7 +135,9 @@ def topic_path(endpoint, topic_consumer) -> str:
 @pytest.mark.asyncio()
 async def topic_with_messages(driver, topic_path):
     pass
-    writer = driver.topic_client.topic_writer(topic_path, producer_and_message_group_id="fixture-producer-id")
+    writer = driver.topic_client.topic_writer(
+        topic_path, producer_and_message_group_id="fixture-producer-id"
+    )
     await writer.write_with_ack(
         ydb.TopicWriterMessage(data="123".encode()),
         ydb.TopicWriterMessage(data="456".encode()),

tests/topics/test_topic_reader.py

@@ -1,12 +1,11 @@
 import pytest
 
-from ydb._topic_reader.topic_reader_asyncio import PublicAsyncIOReader
-from ydb import TopicReaderSettings
-
 
 @pytest.mark.asyncio
 class TestTopicWriterAsyncIO:
-    async def test_read_message(self, driver, topic_path, topic_with_messages, topic_consumer):
+    async def test_read_message(
+        self, driver, topic_path, topic_with_messages, topic_consumer
+    ):
         reader = driver.topic_client.topic_reader(topic_consumer, topic_path)
 
         assert await reader.receive_batch() is not None

ydb/_topic_reader/datatypes.py

@@ -87,4 +87,7 @@ def end_offset(self) -> int:
     @property
     def is_alive(self) -> bool:
         state = self._partition_session.state
-        return state == PartitionSession.State.Active or state == PartitionSession.State.GracefulShutdown
+        return (
+            state == PartitionSession.State.Active
+            or state == PartitionSession.State.GracefulShutdown
+        )

ydb/_topic_reader/topic_reader.py

@@ -1,24 +1,18 @@
-import abc
 import concurrent.futures
 import enum
-import io
 import datetime
 from dataclasses import dataclass
 from typing import (
     Union,
     Optional,
     List,
-    Mapping,
-    Callable,
     Iterable,
-    AsyncIterable,
-    AsyncContextManager,
-    Any, Dict,
 )
 
-from ydb import RetrySettings
-from ydb._topic_wrapper.common import OffsetsRange, TokenGetterFuncType
-from ydb._topic_wrapper.reader import StreamReadMessage
+from ..table import RetrySettings
+from .datatypes import ICommittable, PublicBatch, PublicMessage
+from .._topic_wrapper.common import OffsetsRange, TokenGetterFuncType
+from .._topic_wrapper.reader import StreamReadMessage
 
 
 class Selector:
@@ -47,7 +41,9 @@ async def sessions_stat(self) -> List["SessionStat"]:
         """
         raise NotImplementedError()
 
-    def messages(self, *, timeout: Union[float, None] = None) -> Iterable["PublicMessage"]:
+    def messages(
+        self, *, timeout: Union[float, None] = None
+    ) -> Iterable[PublicMessage]:
         """
         todo?
 
@@ -59,7 +55,7 @@ def messages(self, *, timeout: Union[float, None] = None) -> Iterable["PublicMes
         """
         raise NotImplementedError()
 
-    def receive_message(self, *, timeout: Union[float, None] = None) -> "PublicMessage":
+    def receive_message(self, *, timeout: Union[float, None] = None) -> PublicMessage:
         """
         Block until receive new message
         It has no async_ version for prevent lost messages, use async_wait_message as signal for new batches available.
@@ -85,7 +81,7 @@ def batches(
         max_messages: Union[int, None] = None,
         max_bytes: Union[int, None] = None,
         timeout: Union[float, None] = None,
-    ) -> Iterable["PublicBatch"]:
+    ) -> Iterable[PublicBatch]:
         """
         Block until receive new batch.
         It has no async_ version for prevent lost messages, use async_wait_message as signal for new batches available.
@@ -101,7 +97,7 @@ def receive_batch(
         max_messages: Union[int, None] = None,
         max_bytes: Union[int, None],
         timeout: Union[float, None] = None,
-    ) -> Union["PublicBatch", None]:
+    ) -> Union[PublicBatch, None]:
         """
         Get one messages batch from reader
         It has no async_ version for prevent lost messages, use async_wait_message as signal for new batches available.
@@ -111,7 +107,7 @@ def receive_batch(
         """
         raise NotImplementedError()
 
-    def commit(self, mess: "ICommittable"):
+    def commit(self, mess: ICommittable):
         """
         Put commit message to internal buffer.
 
@@ -121,7 +117,7 @@ def commit(self, mess: "ICommittable"):
         raise NotImplementedError()
 
     def commit_with_ack(
-        self, mess: "ICommittable"
+        self, mess: ICommittable
     ) -> Union["CommitResult", List["CommitResult"]]:
         """
         write commit message to a buffer and wait ack from the server.
@@ -131,7 +127,7 @@ def commit_with_ack(
         raise NotImplementedError()
 
     def async_commit_with_ack(
-        self, mess: "ICommittable"
+        self, mess: ICommittable
     ) -> Union["CommitResult", List["CommitResult"]]:
         """
         write commit message to a buffer and return Future for wait result.
@@ -184,7 +180,7 @@ def _init_message(self) -> StreamReadMessage.InitRequest:
             consumer=self.consumer,
         )
 
-    def _retry_settings(self)->RetrySettings:
+    def _retry_settings(self) -> RetrySettings:
         return RetrySettings(idempotent=True)