Merge pull request #242 clear interface

Author: rekby

examples/topic/reader_async_example.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import time
 
 import ydb
@@ -77,33 +76,19 @@ async def get_one_batch_from_external_loop_async(reader: ydb.TopicReaderAsyncIO)
         await reader.commit(batch)
 
 
-async def auto_deserialize_message(db: ydb.aio.Driver):
-    # async, batch work similar to this
-
-    async with db.topic_client.reader(
-        "/database/topic/path", consumer="asd", deserializer=json.loads
-    ) as reader:
-        while True:
-            message = await reader.receive_message()
-            print(
-                message.data.Name
-            )  # message.data replaces by json.loads(message.data) of raw message
-            reader.commit(message)
-
-
 async def handle_partition_stop(reader: ydb.TopicReaderAsyncIO):
     while True:
         message = await reader.receive_message()
         time.sleep(123)  # some work
-        if message.is_alive:
+        if message.alive:
             time.sleep(1)  # some other work
             await reader.commit(message)
 
 
 async def handle_partition_stop_batch(reader: ydb.TopicReaderAsyncIO):
     def process_batch(batch):
         for message in batch.messages:
-            if not batch.is_alive:
+            if not batch.alive:
                 # no reason work with expired batch
                 # go read next - good batch
                 return
@@ -115,55 +100,5 @@ def process_batch(batch):
         process_batch(batch)
 
 
-async def connect_and_read_few_topics(db: ydb.aio.Driver):
-    with db.topic_client.reader(
-        [
-            "/database/topic/path",
-            ydb.TopicSelector("/database/second-topic", partitions=3),
-        ]
-    ) as reader:
-        while True:
-            message = await reader.receive_message()
-            await _process(message)
-            await reader.commit(message)
-
-
-async def advanced_commit_notify(db: ydb.aio.Driver):
-    def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
-        print(event.topic)
-        print(event.offset)
-
-    async with db.topic_client.reader(
-        "/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit
-    ) as reader:
-        while True:
-            message = await reader.receive_message()
-            await _process(message)
-            await reader.commit(message)
-
-
-async def advanced_read_with_own_progress_storage(db: ydb.TopicReaderAsyncIO):
-    async def on_get_partition_start_offset(
-        req: ydb.TopicReaderEvents.OnPartitionGetStartOffsetRequest,
-    ) -> ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse:
-        # read current progress from database
-        resp = ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse()
-        resp.start_offset = 123
-        return resp
-
-    async with db.topic_client.reader(
-        "/local/test",
-        consumer="consumer",
-        on_get_partition_start_offset=on_get_partition_start_offset,
-    ) as reader:
-        while True:
-            mess = reader.receive_message()
-            await _process(mess)
-            # save progress to own database
-
-            # no commit progress to topic service
-            # reader.commit(mess)
-
-
 async def _process(msg):
     raise NotImplementedError()

examples/topic/reader_example.py

@@ -1,4 +1,3 @@
-import json
 import time
 
 import ydb
@@ -37,14 +36,6 @@ def process_messages_batch_explicit_commit(reader: ydb.TopicReader):
         reader.commit(batch)
 
 
-def process_messages_batch_context_manager_commit(reader: ydb.TopicReader):
-    while True:
-        batch = reader.receive_batch()
-        with reader.commit_on_exit(batch):
-            for message in batch.messages:
-                _process(message)
-
-
 def get_message_with_timeout(reader: ydb.TopicReader):
     try:
         message = reader.receive_message(timeout=1)
@@ -85,33 +76,19 @@ def get_one_batch_from_external_loop(reader: ydb.TopicReader):
         reader.commit(batch)
 
 
-def auto_deserialize_message(db: ydb.Driver):
-    # async, batch work similar to this
-
-    reader = db.topic_client.reader(
-        "/database/topic/path", consumer="asd", deserializer=json.loads
-    )
-    while True:
-        message = reader.receive_message()
-        print(
-            message.data.Name
-        )  # message.data replaces by json.loads(message.data) of raw message
-        reader.commit(message)
-
-
 def handle_partition_stop(reader: ydb.TopicReader):
     while True:
         message = reader.receive_message()
         time.sleep(123)  # some work
-        if message.is_alive:
+        if message.alive:
             time.sleep(1)  # some other work
             reader.commit(message)
 
 
 def handle_partition_stop_batch(reader: ydb.TopicReader):
     def process_batch(batch):
         for message in batch.messages:
-            if not batch.is_alive:
+            if not batch.alive:
                 # no reason work with expired batch
                 # go read next - good batch
                 return
@@ -123,19 +100,6 @@ def process_batch(batch):
         process_batch(batch)
 
 
-def connect_and_read_few_topics(db: ydb.Driver):
-    with db.topic_client.reader(
-        [
-            "/database/topic/path",
-            ydb.TopicSelector("/database/second-topic", partitions=3),
-        ]
-    ) as reader:
-        while True:
-            message = reader.receive_message()
-            _process(message)
-            reader.commit(message)
-
-
 def handle_partition_graceful_stop_batch(reader: ydb.TopicReader):
     # no special handle, but batch will contain less than prefer count messages
     while True:
@@ -144,54 +108,5 @@ def handle_partition_graceful_stop_batch(reader: ydb.TopicReader):
         reader.commit(batch)
 
 
-def advanced_commit_notify(db: ydb.Driver):
-    def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
-        print(event.topic)
-        print(event.offset)
-
-    with db.topic_client.reader(
-        "/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit
-    ) as reader:
-        while True:
-            message = reader.receive_message()
-            with reader.commit_on_exit(message):
-                _process(message)
-
-
-def advanced_read_with_own_progress_storage(db: ydb.TopicReader):
-    def on_get_partition_start_offset(
-        req: ydb.TopicReaderEvents.OnPartitionGetStartOffsetRequest,
-    ) -> ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse:
-
-        # read current progress from database
-        resp = ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse()
-        resp.start_offset = 123
-        return resp
-
-    with db.topic_client.reader(
-        "/local/test",
-        consumer="consumer",
-        on_get_partition_start_offset=on_get_partition_start_offset,
-    ) as reader:
-        while True:
-            mess = reader.receive_message()
-            _process(mess)
-            # save progress to own database
-
-            # no commit progress to topic service
-            # reader.commit(mess)
-
-
-def get_current_statistics(reader: ydb.TopicReader):
-    # sync
-    stat = reader.sessions_stat()
-    print(stat)
-
-    # with feature
-    f = reader.async_sessions_stat()
-    stat = f.result()
-    print(stat)
-
-
 def _process(msg):
     raise NotImplementedError()

examples/topic/writer_async_example.py

@@ -1,6 +1,5 @@
 import asyncio
 import datetime
-import json
 from typing import Dict, List
 
 import ydb
@@ -84,13 +83,6 @@ async def send_messages_with_wait_ack(writer: ydb.TopicWriterAsyncIO):
     await writer.flush()
 
 
-async def send_json_message(db: ydb.aio.Driver):
-    async with db.topic_client.writer(
-        "/database/path/topic", serializer=json.dumps
-    ) as writer:
-        await writer.write({"a": 123})
-
-
 async def send_messages_and_wait_all_commit_with_flush(writer: ydb.TopicWriterAsyncIO):
     for i in range(10):
         await writer.write(ydb.TopicWriterMessage("%s" % i))
@@ -127,8 +119,3 @@ async def switch_messages_with_many_producers(
 
     # all ok, explicit return - for better
     return
-
-
-async def get_current_statistics(reader: ydb.TopicReaderAsyncIO):
-    stat = await reader.sessions_stat()
-    print(stat)

examples/topic/writer_example.py

@@ -1,6 +1,5 @@
 import concurrent.futures
 import datetime
-import json
 from typing import Dict, List
 from concurrent.futures import Future, wait
 
@@ -85,7 +84,10 @@ def send_messages_with_manual_seqno(writer: ydb.TopicWriter):
 def send_messages_with_wait_ack(writer: ydb.TopicWriter):
     # Explicit future wait
     writer.async_write_with_ack(
-        ydb.TopicWriterMessage("mess", seqno=1), ydb.TopicWriterMessage("mess", seqno=2)
+        [
+            ydb.TopicWriterMessage("mess", seqno=1),
+            ydb.TopicWriterMessage("mess", seqno=2),
+        ]
     ).result()
 
     # implicit, by sync call
@@ -102,13 +104,6 @@ def send_messages_with_wait_ack(writer: ydb.TopicWriter):
     writer.flush()
 
 
-def send_json_message(db: ydb.Driver):
-    with db.topic_client.writer(
-        "/database/path/topic", serializer=json.dumps
-    ) as writer:
-        writer.write({"a": 123})
-
-
 def send_messages_and_wait_all_commit_with_flush(writer: ydb.TopicWriter):
     for i in range(10):
         content = "%s" % i

ydb/_topic_reader/datatypes.py

@@ -7,7 +7,7 @@
 from collections import deque
 from dataclasses import dataclass, field
 import datetime
-from typing import Mapping, Union, Any, List, Dict, Deque, Optional
+from typing import Union, Any, List, Dict, Deque, Optional
 
 from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange, Codec
 from ydb._topic_reader import topic_reader_asyncio
@@ -26,7 +26,7 @@ def _commit_get_offsets_range(self) -> OffsetsRange:
 class ISessionAlive(abc.ABC):
     @property
     @abc.abstractmethod
-    def is_alive(self) -> bool:
+    def alive(self) -> bool:
         pass
 
 
@@ -54,8 +54,8 @@ def _commit_get_offsets_range(self) -> OffsetsRange:
 
     # ISessionAlive implementation
     @property
-    def is_alive(self) -> bool:
-        raise NotImplementedError()
+    def alive(self) -> bool:
+        return not self._partition_session.closed
 
 
 @dataclass
@@ -127,16 +127,18 @@ def ack_notify(self, offset: int):
                 break
 
     def close(self):
-        try:
-            self._ensure_not_closed()
-        except topic_reader_asyncio.TopicReaderCommitToExpiredPartition:
+        if self.closed:
             return
 
         self.state = PartitionSession.State.Stopped
         exception = topic_reader_asyncio.TopicReaderCommitToExpiredPartition()
         for waiter in self._ack_waiters:
             waiter._finish_error(exception)
 
+    @property
+    def closed(self):
+        return self.state == PartitionSession.State.Stopped
+
     def _ensure_not_closed(self):
         if self.state == PartitionSession.State.Stopped:
             raise topic_reader_asyncio.TopicReaderCommitToExpiredPartition()
@@ -164,7 +166,6 @@ def _finish_error(self, error: Exception):
 
 @dataclass
 class PublicBatch(ICommittable, ISessionAlive):
-    session_metadata: Mapping[str, str]
     messages: List[PublicMessage]
     _partition_session: PartitionSession
     _bytes_size: int
@@ -184,12 +185,8 @@ def empty(self) -> bool:
 
     # ISessionAlive implementation
     @property
-    def is_alive(self) -> bool:
-        state = self._partition_session.state
-        return (
-            state == PartitionSession.State.Active
-            or state == PartitionSession.State.GracefulShutdown
-        )
+    def alive(self) -> bool:
+        return not self._partition_session.closed
 
     def pop_message(self) -> PublicMessage:
         return self.messages.pop(0)

ydb/_topic_reader/topic_reader.py

@@ -36,19 +36,6 @@ class PublicReaderSettings:
 
     # decoder_executor, must be set for handle non raw messages
     decoder_executor: Optional[concurrent.futures.Executor] = None
-
-    # on_commit: Callable[["Events.OnCommit"], None] = None
-    # on_get_partition_start_offset: Callable[
-    #     ["Events.OnPartitionGetStartOffsetRequest"],
-    #     "Events.OnPartitionGetStartOffsetResponse",
-    # ] = None
-    # on_partition_session_start: Callable[["StubEvent"], None] = None
-    # on_partition_session_stop: Callable[["StubEvent"], None] = None
-    # on_partition_session_close: Callable[["StubEvent"], None] = None  # todo?
-    # deserializer: Union[Callable[[bytes], Any], None] = None
-    # one_attempt_connection_timeout: Union[float, None] = 1
-    # connection_timeout: Union[float, None] = None
-    # retry_policy: Union["RetryPolicy", None] = None
     update_token_interval: Union[int, float] = 3600
 
     def _init_message(self) -> StreamReadMessage.InitRequest: