wip

README.md

@@ -34,7 +34,7 @@ pip install streamstore
 
 ## Examples
 
-`examples/streaming` directory in the [repo](https://github.com/s2-streamstore/s2-sdk-python/tree/main/examples/streaming) contain examples for streaming APIs.
+`examples/` directory in the [repo](https://github.com/s2-streamstore/s2-sdk-python/tree/main/examples/) contain examples for streaming APIs.
 
 ## Get in touch
 

examples/streaming/producer.py → examples/append_session.py

@@ -1,15 +1,17 @@
 import asyncio
 import os
 import random
+from typing import AsyncIterable
+
 from streamstore import S2
-from streamstore.schemas import Record, AppendInput
+from streamstore.schemas import AppendInput, Record
 
 AUTH_TOKEN = os.getenv("S2_AUTH_TOKEN")
 MY_BASIN = os.getenv("MY_BASIN")
 MY_STREAM = os.getenv("MY_STREAM")
 
 
-async def append_inputs():
+async def append_inputs() -> AsyncIterable[AppendInput]:
     num_inputs = random.randint(1, 100)
     for _ in range(num_inputs):
         num_records = random.randint(1, 100)

examples/append_session_with_auto_batching.py

@@ -0,0 +1,37 @@
+import asyncio
+import os
+import random
+from datetime import timedelta
+from typing import AsyncIterable
+
+from streamstore import S2
+from streamstore.schemas import Record
+
+AUTH_TOKEN = os.getenv("S2_AUTH_TOKEN")
+MY_BASIN = os.getenv("MY_BASIN")
+MY_STREAM = os.getenv("MY_STREAM")
+
+
+async def records() -> AsyncIterable[Record]:
+    num_records = random.randint(1, 1000)
+    for _ in range(num_records):
+        body_size = random.randint(1, 1000)
+        if random.random() < 0.5:
+            await asyncio.sleep(random.random() * 2.5)
+        yield Record(body=os.urandom(body_size))
+
+
+async def producer():
+    async with S2(auth_token=AUTH_TOKEN) as s2:
+        stream = s2[MY_BASIN][MY_STREAM]
+        async for output in stream.append_session_with_auto_batching(
+            records=records(),
+            max_records_per_batch=10,
+            max_linger_per_batch=timedelta(milliseconds=5),
+        ):
+            num_appended_records = output.end_seq_num - output.start_seq_num
+            print(f"appended {num_appended_records} records")
+
+
+if __name__ == "__main__":
+    asyncio.run(producer())

examples/streaming/consumer.py → examples/read_session.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+
 from streamstore import S2
 
 AUTH_TOKEN = os.getenv("S2_AUTH_TOKEN")