fix examples

Author: rekby

examples/topic/reader_async_example.py

@@ -10,12 +10,12 @@ async def connect():
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    reader = ydb.TopicClientAsyncIO(db).reader("/local/topic", consumer="consumer")
+    reader = db.topic_client.reader("/local/topic", consumer="consumer")
     return reader
 
 
 async def create_reader_and_close_with_context_manager(db: ydb.aio.Driver):
-    async with ydb.TopicClientAsyncIO(db).reader(
+    async with db.topic_client.reader(
         "/database/topic/path", consumer="consumer"
     ) as reader:
         ...
@@ -80,7 +80,7 @@ async def get_one_batch_from_external_loop_async(reader: ydb.TopicReaderAsyncIO)
 async def auto_deserialize_message(db: ydb.aio.Driver):
     # async, batch work similar to this
 
-    async with ydb.TopicClientAsyncIO(db).reader(
+    async with db.topic_client.reader(
         "/database/topic/path", consumer="asd", deserializer=json.loads
     ) as reader:
         while True:
@@ -116,7 +116,7 @@ def process_batch(batch):
 
 
 async def connect_and_read_few_topics(db: ydb.aio.Driver):
-    with ydb.TopicClientAsyncIO(db).reader(
+    with db.topic_client.reader(
         [
             "/database/topic/path",
             ydb.TopicSelector("/database/second-topic", partitions=3),
@@ -133,7 +133,7 @@ def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
         print(event.topic)
         print(event.offset)
 
-    async with ydb.TopicClientAsyncIO(db).reader(
+    async with db.topic_client.reader(
         "/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit
     ) as reader:
         while True:
@@ -151,12 +151,13 @@ async def on_get_partition_start_offset(
         resp.start_offset = 123
         return resp
 
-    async with ydb.TopicClient(db).reader(
+    async with db.topic_client.reader(
         "/local/test",
         consumer="consumer",
         on_get_partition_start_offset=on_get_partition_start_offset,
     ) as reader:
-        async for mess in reader.messages():
+        while True:
+            mess = reader.receive_message()
             await _process(mess)
             # save progress to own database
 

examples/topic/reader_example.py

@@ -9,33 +9,37 @@ def connect():
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    reader = ydb.TopicClient(db).reader("/local/topic", consumer="consumer")
+    reader = db.topic_client.reader("/local/topic", consumer="consumer")
     return reader
 
 
 def create_reader_and_close_with_context_manager(db: ydb.Driver):
-    with ydb.TopicClient(db).reader(
+    with db.topic_client.reader(
         "/database/topic/path", consumer="consumer", buffer_size_bytes=123
     ) as reader:
-        for message in reader:
+        while True:
+            message = reader.receive_message()
             pass
 
 
 def print_message_content(reader: ydb.TopicReader):
-    for message in reader.messages():
+    while True:
+        message = reader.receive_message()
         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):
+    while True:
+        batch = reader.receive_batch()
         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):
+    while True:
+        batch = reader.receive_batch()
         with reader.commit_on_exit(batch):
             for message in batch.messages:
                 _process(message)
@@ -52,9 +56,12 @@ def get_message_with_timeout(reader: ydb.TopicReader):
 
 
 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")
+    while True:
+        try:
+            message = reader.receive_message(timeout=1)
+            _process(message)
+        except TimeoutError:
+            print("Have no new messages in a second")
 
 
 def get_a_message_from_external_loop(reader: ydb.TopicReader):
@@ -81,30 +88,23 @@ def get_one_batch_from_external_loop(reader: ydb.TopicReader):
 def auto_deserialize_message(db: ydb.Driver):
     # async, batch work similar to this
 
-    reader = ydb.TopicClient(db).reader(
+    reader = db.topic_client.reader(
         "/database/topic/path", consumer="asd", deserializer=json.loads
     )
-    for message in reader.messages():
+    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 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
+    while True:
+        message = reader.receive_message()
+        time.sleep(123)  # some work
         if message.is_alive:
-            time.sleep(123)  # some other work
+            time.sleep(1)  # some other work
             reader.commit(message)
 
 
@@ -118,25 +118,28 @@ def process_batch(batch):
             _process(message)
         reader.commit(batch)
 
-    for batch in reader.batches():
+    while True:
+        batch = reader.receive_batch()
         process_batch(batch)
 
 
 def connect_and_read_few_topics(db: ydb.Driver):
-    with ydb.TopicClient(db).reader(
+    with db.topic_client.reader(
         [
             "/database/topic/path",
             ydb.TopicSelector("/database/second-topic", partitions=3),
         ]
     ) as reader:
-        for message in 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
-    for batch in reader.batches():
+    while True:
+        batch = reader.receive_batch()
         _process(batch)
         reader.commit(batch)
 
@@ -146,10 +149,11 @@ def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
         print(event.topic)
         print(event.offset)
 
-    with ydb.TopicClient(db).reader(
+    with db.topic_client.reader(
         "/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit
     ) as reader:
-        for message in reader:
+        while True:
+            message = reader.receive_message()
             with reader.commit_on_exit(message):
                 _process(message)
 
@@ -164,12 +168,13 @@ def on_get_partition_start_offset(
         resp.start_offset = 123
         return resp
 
-    with ydb.TopicClient(db).reader(
+    with db.topic_client.reader(
         "/local/test",
         consumer="consumer",
         on_get_partition_start_offset=on_get_partition_start_offset,
     ) as reader:
-        for mess in reader:
+        while True:
+            mess = reader.receive_message()
             _process(mess)
             # save progress to own database
 

examples/topic/writer_async_example.py

@@ -1,4 +1,5 @@
 import asyncio
+import datetime
 import json
 import time
 from typing import Dict, List
@@ -8,23 +9,24 @@
 
 
 async def create_writer(db: ydb.aio.Driver):
-    async with ydb.TopicClientAsyncIO(db).writer(
+    async with db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     ) as writer:
         await writer.write(TopicWriterMessage("asd"))
 
 
 async def connect_and_wait(db: ydb.aio.Driver):
-    async with ydb.TopicClientAsyncIO(db).writer(
+    async with db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     ) as writer:
-        writer.wait_init()
+        info = await writer.wait_init()
+        ...
 
 
 async def connect_without_context_manager(db: ydb.aio.Driver):
-    writer = ydb.TopicClientAsyncIO(db).writer(
+    writer = db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     )
@@ -49,7 +51,7 @@ async def send_messages(writer: ydb.TopicWriterAsyncIO):
 
     # with meta
     await writer.write(
-        ydb.TopicWriterMessage("asd", seqno=123, created_at_ns=time.time_ns())
+        ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now())
     )
 
 
@@ -71,7 +73,7 @@ async def send_messages_with_manual_seqno(writer: ydb.TopicWriter):
 
 async def send_messages_with_wait_ack(writer: ydb.TopicWriterAsyncIO):
     # future wait
-    await writer.write_with_result(
+    await writer.write_with_ack(
         [
             ydb.TopicWriterMessage("mess", seqno=1),
             ydb.TopicWriterMessage("mess", seqno=2),
@@ -84,10 +86,10 @@ async def send_messages_with_wait_ack(writer: ydb.TopicWriterAsyncIO):
 
 
 async def send_json_message(db: ydb.aio.Driver):
-    async with ydb.TopicClientAsyncIO(db).writer(
+    async with db.topic_client.writer(
         "/database/path/topic", serializer=json.dumps
     ) as writer:
-        writer.write({"a": 123})
+        await writer.write({"a": 123})
 
 
 async def send_messages_and_wait_all_commit_with_flush(writer: ydb.TopicWriterAsyncIO):
@@ -99,14 +101,11 @@ async def send_messages_and_wait_all_commit_with_flush(writer: ydb.TopicWriterAs
 async def send_messages_and_wait_all_commit_with_results(
     writer: ydb.TopicWriterAsyncIO,
 ):
-    last_future = None
     for i in range(10):
         content = "%s" % i
-        last_future = await writer.write_with_ack(content)
+        last_future = await writer.write(content)
 
-    await asyncio.wait(last_future)
-    if last_future.exception() is not None:
-        raise last_future.exception()
+    await writer.flush()
 
 
 async def switch_messages_with_many_producers(
@@ -118,7 +117,7 @@ async def switch_messages_with_many_producers(
         # select writer for the msg
         writer_idx = msg[:1]
         writer = writers[writer_idx]
-        future = await writer.write_with_ack(msg)
+        future = await writer.write_with_ack_future(msg)
         futures.append(future)
 
     # wait acks from all writes

examples/topic/writer_example.py

@@ -1,4 +1,5 @@
 import concurrent.futures
+import datetime
 import json
 import time
 from typing import Dict, List
@@ -8,36 +9,36 @@
 from ydb import TopicWriterMessage
 
 
-async def connect():
-    db = ydb.aio.Driver(
+def connect():
+    db = ydb.Driver(
         connection_string="grpc://localhost:2135?database=/local",
         credentials=ydb.credentials.AnonymousCredentials(),
     )
-    writer = ydb.TopicClientAsyncIO(db).writer(
+    writer = db.topic_client.writer(
         "/local/topic",
         producer_id="producer-id",
     )
-    await writer.write(TopicWriterMessage("asd"))
+    writer.write(TopicWriterMessage("asd"))
 
 
 def create_writer(db: ydb.Driver):
-    with ydb.TopicClient(db).writer(
+    with db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     ) as writer:
         writer.write(TopicWriterMessage("asd"))
 
 
 def connect_and_wait(db: ydb.Driver):
-    with ydb.TopicClient(db).writer(
+    with db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     ) as writer:
-        writer.wait()
+        info = writer.wait_init()
 
 
 def connect_without_context_manager(db: ydb.Driver):
-    writer = ydb.TopicClient(db).writer(
+    writer = db.topic_client.writer(
         "/database/topic/path",
         producer_id="producer-id",
     )
@@ -61,7 +62,7 @@ def send_messages(writer: ydb.TopicWriter):
     )  # send few messages by one call
 
     # with meta
-    writer.write(ydb.TopicWriterMessage("asd", seqno=123, created_at_ns=time.time_ns()))
+    writer.write(ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now()))
 
 
 def send_message_without_block_if_internal_buffer_is_full(
@@ -101,7 +102,7 @@ def send_messages_with_wait_ack(writer: ydb.TopicWriter):
 
 
 def send_json_message(db: ydb.Driver):
-    with ydb.TopicClient(db).writer(
+    with db.topic_client.writer(
         "/database/path/topic", serializer=json.dumps
     ) as writer:
         writer.write({"a": 123})