initial topic writer

Author: rekby

.gitignore

@@ -3,4 +3,5 @@ ydb.egg-info/
 /.idea
 /tox
 /venv
-/ydb_certs
+/ydb_certs
+/tmp

AUTHORS

@@ -1,4 +1,5 @@
-The following authors have created the source code of "YDB Python SDK"
+The following authors have created the source code of "Yandex Database Python SDK"
 published and distributed by YANDEX LLC as the owner:
 
 Vitalii Gridnev <[email protected]>
+Timofey Koolin <[email protected]>

examples/topic/reader_async_example.py

@@ -0,0 +1,172 @@
+import asyncio
+import json
+import time
+
+import ydb
+
+
+async def connect():
+    db = ydb.aio.Driver(connection_string="grpc://localhost:2135?database=/local", credentials=ydb.credentials.AnonymousCredentials())
+    reader = ydb.TopicClientAsyncIO(db).topic_reader("/local/topic", consumer="consumer")
+
+
+async def create_reader_and_close_with_context_manager(db: ydb.aio.Driver):
+    with ydb.TopicClientAsyncIO(db).topic_reader("/database/topic/path", consumer="consumer") as reader:
+        async for message in reader.messages():
+            pass
+
+
+async def print_message_content(reader: ydb.TopicReaderAsyncIO):
+    async for message in reader.messages():
+        print("text", message.data.read().decode("utf-8"))
+        # await and async_commit need only for sync commit mode - for wait ack from servr
+        await reader.commit(message)
+
+
+async def process_messages_batch_explicit_commit(reader: ydb.TopicReaderAsyncIO):
+    # Explicit commit example
+    async for batch in reader.batches(max_messages=100, timeout=2):
+        async with asyncio.TaskGroup() as tg:
+            for message in batch.messages:
+                tg.create_task(_process(message))
+
+        # wait complete of process all messages from batch be taskgroup context manager
+        # and commit complete batch
+        await reader.commit(batch)
+
+
+async def process_messages_batch_context_manager_commit(reader: ydb.TopicReaderAsyncIO):
+    # Commit with context manager
+    async for batch in reader.batches():
+        async with reader.commit_on_exit(batch), asyncio.TaskGroup() as tg:
+            for message in batch.messages:
+                tg.create_task(_process(message))
+
+
+async def get_message_with_timeout(reader: ydb.TopicReaderAsyncIO):
+    try:
+        message = await asyncio.wait_for(reader.receive_message(), timeout=1)
+    except TimeoutError:
+        print("Have no new messages in a second")
+        return
+
+    print("mess", message.data)
+
+
+async def get_all_messages_with_small_wait(reader: ydb.TopicReaderAsyncIO):
+    async for message in reader.messages(timeout=1):
+        await _process(message)
+    print("Have no new messages in a second")
+
+
+async def get_a_message_from_external_loop(reader: ydb.TopicReaderAsyncIO):
+    for i in range(10):
+        try:
+            message = await asyncio.wait_for(reader.receive_message(), timeout=1)
+        except TimeoutError:
+            return
+        await _process(message)
+
+
+async def get_one_batch_from_external_loop_async(reader: ydb.TopicReaderAsyncIO):
+    for i in range(10):
+        try:
+            batch = await asyncio.wait_for(reader.receive_batch(), timeout=2)
+        except TimeoutError:
+            return
+
+        for message in batch.messages:
+            await _process(message)
+        await reader.commit(batch)
+
+
+async def auto_deserialize_message(db: ydb.aio.Driver):
+    # async, batch work similar to this
+
+    async with ydb.TopicClientAsyncIO(db).topic_reader("/database/topic/path", consumer="asd", deserializer=json.loads) as reader:
+        async for message in reader.messages():
+            print(message.data.Name)  # message.data replaces by json.loads(message.data) of raw message
+            reader.commit(message)
+
+
+async def commit_batch_with_context(reader: ydb.TopicReaderAsyncIO):
+    async for batch in reader.batches():
+        async with reader.commit_on_exit(batch):
+            for message in batch.messages:
+                if not batch.is_alive:
+                    break
+                await _process(message)
+
+
+async def handle_partition_stop(reader: ydb.TopicReaderAsyncIO):
+    async for message in reader.messages():
+        time.sleep(1)  # some work
+        if message.is_alive:
+            time.sleep(123)  # 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:
+                # no reason work with expired batch
+                # go read next - good batch
+                return
+            await _process(message)
+        await reader.commit(batch)
+
+    async for batch in reader.batches():
+        process_batch(batch)
+
+
+async def connect_and_read_few_topics(db: ydb.aio.Driver):
+    with ydb.TopicClientAsyncIO(db).topic_reader(
+            ["/database/topic/path", ydb.TopicSelector("/database/second-topic", partitions=3)]) as reader:
+        async for message in reader.messages():
+            await _process(message)
+            await reader.commit(message)
+
+
+async def handle_partition_graceful_stop_batch(reader: ydb.TopicReaderAsyncIO):
+    # no special handle, but batch will contain less than prefer count messages
+    async for batch in reader.batches():
+        await _process(batch)
+        reader.commit(batch)
+
+
+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 ydb.TopicClientAsyncIO(db).topic_reader("/local",
+                                                       consumer="consumer",
+                                                       commit_batch_time=4,
+                                                       on_commit=on_commit) as reader:
+        async for message in reader.messages():
+            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 ydb.TopicClient(db).topic_reader("/local/test", consumer="consumer",
+                                                    on_get_partition_start_offset=on_get_partition_start_offset
+                                                ) as reader:
+        async for mess in reader.messages():
+            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

@@ -0,0 +1,173 @@
+import json
+import time
+
+import ydb
+
+
+def connect():
+    db = ydb.Driver(connection_string="grpc://localhost:2135?database=/local", credentials=ydb.credentials.AnonymousCredentials())
+    reader = ydb.TopicClient(db).topic_reader("/local/topic", consumer="consumer")
+
+
+def create_reader_and_close_with_context_manager(db: ydb.Driver):
+    with ydb.TopicClient(db).topic_reader("/database/topic/path", consumer="consumer", buffer_size_bytes=123) as reader:
+        for message in reader:
+            pass
+
+
+def print_message_content(reader: ydb.TopicReader):
+    for message in reader.messages():
+        print("text", message.data.read().decode("utf-8"))
+        reader.commit(message)
+
+
+def process_messages_batch_explicit_commit(reader: ydb.TopicReader):
+    for batch in reader.batches(max_messages=100, timeout=2):
+        for message in batch.messages:
+            _process(message)
+        reader.commit(batch)
+
+
+def process_messages_batch_context_manager_commit(reader: ydb.TopicReader):
+    for batch in reader.batches(max_messages=100, timeout=2):
+        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)
+    except TimeoutError:
+        print("Have no new messages in a second")
+        return
+
+    print("mess", message.data)
+
+
+def get_all_messages_with_small_wait(reader: ydb.TopicReader):
+    for message in reader.messages(timeout=1):
+        _process(message)
+    print("Have no new messages in a second")
+
+
+def get_a_message_from_external_loop(reader: ydb.TopicReader):
+    for i in range(10):
+        try:
+            message = reader.receive_message(timeout=1)
+        except TimeoutError:
+            return
+        _process(message)
+
+
+def get_one_batch_from_external_loop(reader: ydb.TopicReader):
+    for i in range(10):
+        try:
+            batch = reader.receive_batch(timeout=2)
+        except TimeoutError:
+            return
+
+        for message in batch.messages:
+            _process(message)
+        reader.commit(batch)
+
+
+def auto_deserialize_message(db: ydb.Driver):
+    # async, batch work similar to this
+
+    reader = ydb.TopicClient(db).topic_reader("/database/topic/path", consumer="asd", deserializer=json.loads)
+    for message in reader.messages():
+        print(message.data.Name)  # message.data replaces by json.loads(message.data) of raw message
+        reader.commit(message)
+
+
+def commit_batch_with_context(reader: ydb.TopicReader):
+    for batch in reader.batches():
+        with reader.commit_on_exit(batch):
+            for message in batch.messages:
+                if not batch.is_alive:
+                    break
+                _process(message)
+
+
+def handle_partition_stop(reader: ydb.TopicReader):
+    for message in reader.messages():
+        time.sleep(1)  # some work
+        if message.is_alive:
+            time.sleep(123)  # 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:
+                # no reason work with expired batch
+                # go read next - good batch
+                return
+            _process(message)
+        reader.commit(batch)
+
+    for batch in reader.batches():
+        process_batch(batch)
+
+
+def connect_and_read_few_topics(db: ydb.Driver):
+    with ydb.TopicClient(db).topic_reader(["/database/topic/path", ydb.TopicSelector("/database/second-topic", partitions=3)]) as reader:
+        for message in reader:
+            _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
+    for batch in reader.batches():
+        _process(batch)
+        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 ydb.TopicClient(db).topic_reader("/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit) as reader:
+        for message in reader:
+            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 ydb.TopicClient(db).topic_reader("/local/test", consumer="consumer",
+                                              on_get_partition_start_offset=on_get_partition_start_offset
+                                          ) as reader:
+        for mess in reader:
+            _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()
+