fix flake 8 linter

Author: rekby

examples/topic/reader_async_example.py

@@ -13,6 +13,7 @@ async def connect():
     reader = ydb.TopicClientAsyncIO(db).topic_reader(
         "/local/topic", consumer="consumer"
     )
+    return reader
 
 
 async def create_reader_and_close_with_context_manager(db: ydb.aio.Driver):

examples/topic/reader_example.py

@@ -10,6 +10,7 @@ def connect():
         credentials=ydb.credentials.AnonymousCredentials(),
     )
     reader = ydb.TopicClient(db).topic_reader("/local/topic", consumer="consumer")
+    return reader
 
 
 def create_reader_and_close_with_context_manager(db: ydb.Driver):

examples/topic/writer_async_example.py

@@ -1,17 +1,18 @@
 import asyncio
 import json
 import time
-from typing import Dict, List, Set
+from typing import Dict, List
 
 import ydb
+from ydb import TopicWriterMessage
 
 
 async def create_writer(db: ydb.aio.Driver):
     async with ydb.TopicClientAsyncIO(db).topic_writer(
         "/database/topic/path",
         producer_and_message_group_id="producer-id",
     ) as writer:
-        pass
+        await writer.write(TopicWriterMessage("asd"))
 
 
 async def connect_and_wait(db: ydb.aio.Driver):

examples/topic/writer_example.py

@@ -5,25 +5,27 @@
 from concurrent.futures import Future, wait
 
 import ydb
+from ydb import TopicWriterMessage
 
 
 async def connect():
     db = ydb.aio.Driver(
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    reader = ydb.TopicClientAsyncIO(db).topic_writer(
+    writer = ydb.TopicClientAsyncIO(db).topic_writer(
         "/local/topic",
         producer_and_message_group_id="producer-id",
     )
+    await writer.write(TopicWriterMessage("asd"))
 
 
 def create_writer(db: ydb.Driver):
     with ydb.TopicClient(db).topic_writer(
         "/database/topic/path",
         producer_and_message_group_id="producer-id",
     ) as writer:
-        pass
+        writer.write(TopicWriterMessage("asd"))
 
 
 def connect_and_wait(db: ydb.Driver):

ydb/_topic_reader/__init__.py

@@ -1 +0,0 @@
-from .topic_reader import *

ydb/_topic_reader/topic_reader.py

@@ -253,9 +253,9 @@ def __init__(
         *,
         consumer: str,
         buffer_size_bytes: int = 50 * 1024 * 1024,
-        on_commit: Callable[["OnCommitEvent"], None] = None,
+        on_commit: Callable[["Events.OnCommit"], None] = None,
         on_get_partition_start_offset: Callable[
-            ["OnPartitionGetStartOffsetRequest"], "OnPartitionGetStartOffsetResponse"
+            ["Events.OnPartitionGetStartOffsetRequest"], "Events.OnPartitionGetStartOffsetResponse"
         ] = None,
         on_partition_session_start: Callable[["StubEvent"], None] = None,
         on_partition_session_stop: Callable[["StubEvent"], None] = None,
@@ -390,3 +390,7 @@ class SessionStat:
 class OffsetRange:
     start: int
     end: int
+
+
+class StubEvent:
+    pass

ydb/_topic_wrapper/common.py

@@ -1,6 +1,5 @@
 import abc
 import asyncio
-import queue
 import typing
 from dataclasses import dataclass
 from enum import Enum

ydb/_topic_wrapper/reader.py

@@ -1,12 +1,7 @@
-import abc
 import datetime
-import typing
-from codecs import Codec
 from dataclasses import dataclass, field
 from typing import List, Union, Dict
 
-from google.protobuf.message import Message
-
 from ydb._topic_wrapper.common import OffsetsRange
 
 
@@ -53,7 +48,7 @@ class MessageData:
 
         @dataclass
         class Batch:
-            message_data: List["MessageData"]
+            message_data: List["StreamReadMessage.ReadResponse.MessageData"]
             producer_id: str
             write_session_meta: Dict[str, str]
             codec: int
@@ -62,7 +57,7 @@ class Batch:
         @dataclass
         class PartitionData:
             partition_session_id: int
-            batches: List["Batch"]
+            batches: List["StreamReadMessage.ReadResponse.Batch"]
 
     @dataclass
     class CommitOffsetRequest:
@@ -95,7 +90,7 @@ class PartitionSessionStatusResponse:
 
     @dataclass
     class StartPartitionSessionRequest:
-        partition_session: "PartitionSession"
+        partition_session: "StreamReadMessage.PartitionSession"
         committed_offset: int
         partition_offsets: OffsetsRange
 

ydb/_topic_wrapper/writer.py

@@ -1,4 +1,3 @@
-import asyncio
 import datetime
 import enum
 import typing

ydb/_topic_writer/topic_writer.py

@@ -1,8 +1,6 @@
-import asyncio
 import concurrent.futures
 import datetime
 import enum
-import time
 from dataclasses import dataclass
 from enum import Enum
 from typing import List, Union, TextIO, BinaryIO, Optional, Callable, Mapping, Any, Dict
@@ -17,7 +15,7 @@
 class Writer:
     @property
     def last_seqno(self) -> int:
-        raise NotImplemented()
+        raise NotImplementedError()
 
     def __init__(self, db: ydb.Driver):
         pass
@@ -31,7 +29,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def close(self):
         pass
 
-    MessageType = typing.Union["PublicMessage", "Message.SimpleMessageSourceType"]
+    MessageType = typing.Union["PublicMessage", "PublicMessage.SimpleMessageSourceType"]
 
     def write(
         self,