wip

Author: quettabit

.github/workflows/ci.yml

@@ -1,23 +1,39 @@
 name: CI
-on: 
+
+# NOTE: uv version is also pinned in s2-lite-integration-tests sdks config below.
+env:
+  UV_VERSION: "0.11.3"
+
+on:
   pull_request:
-    types: [opened, edited, synchronize, labeled, unlabeled, ready_for_review, reopened]
+    types:
+      [
+        opened,
+        edited,
+        synchronize,
+        labeled,
+        unlabeled,
+        ready_for_review,
+        reopened,
+      ]
 jobs:
-  ci:
-    name: CI
+  local-checks:
+    name: Local Checks (code quality, unit tests, docs build, PR title)
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
       - name: Install uv
-        uses: astral-sh/setup-uv@v6
+        uses: astral-sh/setup-uv@v7
         with:
-          version: "0.8.2"
+          version: ${{ env.UV_VERSION }}
       - name: Sync dependencies
         run: |
           uv sync --all-groups
       - name: Static code check
         run: uv run poe ci_checker
+      - name: Unit tests
+        run: uv run pytest tests/ -v -m 'not (account or basin or stream or metrics)'
       - name: Check docs build
         working-directory: ./docs
         run: |
@@ -26,20 +42,60 @@ jobs:
         uses: actions/github-script@v7
         with:
           script: |
-              const title = context.payload.pull_request.title;
-              const labels = context.payload.pull_request.labels.map(l => l.name);
-              if (labels.includes('dev')) {
-                const regex = /^(?!feat|fix|refactor|docs|perf|style|test|chore|revert)[a-z].*$/;
-                if (!regex.test(title)) {
-                  core.setFailed(
-                    `PR title "${title}" does not match the commit format for non-user-facing changes`
-                  );
-                }
-              } else {
-                const regex = /^(feat|fix|refactor|docs|perf|style|test|chore|revert)!?:[ ][a-z].*$/;
-                if (!regex.test(title)) {
-                  core.setFailed(
-                    `PR title "${title}" does not match the expected conventional commit format for user-facing changes`
-                  );
-                }
+            const title = context.payload.pull_request.title;
+            const labels = context.payload.pull_request.labels.map(l => l.name);
+            if (labels.includes('dev')) {
+              const regex = /^(?!feat|fix|refactor|docs|perf|style|test|chore|revert)[a-z].*$/;
+              if (!regex.test(title)) {
+                core.setFailed(
+                  `PR title "${title}" does not match the commit format for non-user-facing changes`
+                );
               }
+            } else {
+              const regex = /^(feat|fix|refactor|docs|perf|style|test|chore|revert)!?:[ ][a-z].*$/;
+              if (!regex.test(title)) {
+                core.setFailed(
+                  `PR title "${title}" does not match the expected conventional commit format for user-facing changes`
+                );
+              }
+            }
+
+  s2-cloud-integration-tests:
+    name: S2 Cloud Integration Tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+        with:
+          version: ${{ env.UV_VERSION }}
+      - name: Sync dependencies
+        run: uv sync --group test
+      - name: Run integration tests
+        env:
+          S2_ACCESS_TOKEN: ${{ secrets.S2_ACCESS_TOKEN }}
+        run: uv run pytest tests/ -v -s -m 'account or basin or stream or metrics'
+
+  build-s2-lite:
+    uses: s2-streamstore/s2/.github/workflows/build-s2-lite.yml@main
+
+  s2-lite-integration-tests:
+    name: S2-Lite Integration Tests
+    needs: build-s2-lite
+    uses: s2-streamstore/s2/.github/workflows/sdk-tests.yml@main
+    with:
+      server-binary: server
+      server-args: "--port 8080"
+      server-port: 8080
+      sdks: |
+        [
+          {
+            "name": "python",
+            "repo": "${{ github.repository }}",
+            "ref": "${{ github.ref }}",
+            "lang": "python",
+            "uv-version": "0.11.3",
+            "test_cmd": "uv run pytest tests/ -v -s -m 'account or basin or stream'"
+          }
+        ]

.github/workflows/release.yml

@@ -1,4 +1,4 @@
-name: Release streamstore package
+name: Release s2-sdk package
 on:
   push:
     tags: ["[0-9]+.[0-9]+.[0-9]+*"]

README.md

@@ -1,23 +1,23 @@
-# streamstore
+# s2-sdk
 <div>
   <p>
     <!-- PyPI -->
-    <a href="https://pypi.org/project/streamstore/"><img src="https://img.shields.io/pypi/v/streamstore" /></a>
+    <a href="https://pypi.org/project/s2-sdk/"><img src="https://img.shields.io/pypi/v/s2-sdk" /></a>
     <!-- Read the docs -->
-    <a href="https://streamstore.readthedocs.io/"><img src="https://img.shields.io/readthedocs/streamstore/latest" /></a>
+    <a href="https://s2-sdk.readthedocs.io/"><img src="https://img.shields.io/readthedocs/s2-sdk/latest" /></a>
     <!-- Discord -->
     <a href="https://discord.gg/vTCs7kMkAf"><img src="https://img.shields.io/discord/1209937852528599092?logo=discord" /></a>
     <!-- LICENSE -->
     <a href="https://github.com/s2-streamstore/s2-sdk-python/blob/main/LICENSE"><img src="https://img.shields.io/github/license/s2-streamstore/s2-sdk-python" /></a>
   </p>
 </div>
 
-`streamstore` is the Python package that provides an async client for interacting with [s2.dev](https://s2.dev/).
+`s2_sdk` is the Python package that provides an async client for interacting with [s2.dev](https://s2.dev/).
 
 ## Project links
 
-- [PyPI](https://pypi.org/project/streamstore/)
-- [Documentation](https://streamstore.readthedocs.io/)
+- [PyPI](https://pypi.org/project/s2-sdk/)
+- [Documentation](https://s2-sdk.readthedocs.io/)
 - [GitHub](https://github.com/s2-streamstore/s2-sdk-python)
 
 ## Requirements
@@ -26,10 +26,10 @@ Python >= 3.11
 
 ## Installation
 
-You can install the package from the [Python Package Index](https://pypi.org/project/streamstore) using the package manager of your choice. E.g., with `pip`:
+You can install the package from the [Python Package Index](https://pypi.org/project/s2-sdk) using the package manager of your choice. E.g., with `pip`:
 
 ```bash
-pip install streamstore
+pip install s2-sdk
 ```
 
 ## Examples

docs/source/api-reference.md

@@ -1,36 +1,7 @@
 # API Reference
 
 ```{eval-rst}
-.. module:: streamstore
-
-.. autoclass:: S2
-    :members:
-    :member-order: bysource
-
-.. autoclass:: Basin()
-    :members:
-    :member-order: bysource
-
-.. autoclass:: Stream()
-    :members:
-    :member-order: bysource
-
-.. module:: streamstore.schemas
-    :no-index:
-.. autoclass:: Record(body: bytes, headers: list[tuple[bytes, bytes]] = [])
-    :members:
-
-.. automodule:: streamstore.schemas
-    :members:
-    :exclude-members: Record, Endpoints
-    :member-order: bysource
-
-.. module:: streamstore.schemas
-    :no-index:
-.. autoclass:: Endpoints()
-    :members:
-
-.. automodule:: streamstore.utils
+.. automodule:: s2_sdk
     :members:
     :member-order: bysource
 ```

docs/source/conf.py

@@ -8,9 +8,9 @@
 
 from datetime import date
 
-project = "streamstore"
+project = "s2-sdk"
 copyright = f"{date.today().year}, Bandar Systems Inc"
-release = "5.0.0"
+release = "0.1.0"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

examples/append_session.py

@@ -1,36 +1,31 @@
 import asyncio
 import os
 import random
-from typing import AsyncIterable
 
-from streamstore import S2
-from streamstore.schemas import AppendInput, Record
+from s2_sdk import S2, AppendInput, Record
 
 ACCESS_TOKEN = os.getenv("S2_ACCESS_TOKEN")
 MY_BASIN = os.getenv("MY_BASIN")
 MY_STREAM = os.getenv("MY_STREAM")
 
 
-async def append_inputs_gen() -> AsyncIterable[AppendInput]:
-    num_inputs = random.randint(1, 100)
-    for _ in range(num_inputs):
-        num_records = random.randint(1, 100)
-        records = []
-        for _ in range(num_records):
-            body_size = random.randint(1, 1024)
-            records.append(Record(body=os.urandom(body_size)))
-        input = AppendInput(records)
-        if random.random() < 0.5:
-            await asyncio.sleep(random.random() * 2.5)
-        yield input
-
-
 async def producer():
-    async with S2(access_token=ACCESS_TOKEN) as s2:
+    async with S2(ACCESS_TOKEN) as s2:
         stream = s2[MY_BASIN][MY_STREAM]
-        async for output in stream.append_session(append_inputs_gen()):
-            num_appended_records = output.end_seq_num - output.start_seq_num
-            print(f"appended {num_appended_records} records")
+        async with stream.append_session() as session:
+            num_inputs = random.randint(1, 100)
+            for _ in range(num_inputs):
+                num_records = random.randint(1, 100)
+                records = []
+                for _ in range(num_records):
+                    body_size = random.randint(1, 1024)
+                    records.append(Record(body=os.urandom(body_size)))
+                ticket = await session.submit(AppendInput(records))
+                ack = await ticket
+                num_appended_records = ack.end.seq_num - ack.start.seq_num
+                print(f"appended {num_appended_records} records")
+                if random.random() < 0.5:
+                    await asyncio.sleep(random.random() * 2.5)
 
 
 if __name__ == "__main__":

examples/append_session_with_auto_batching.py

@@ -4,9 +4,7 @@
 from datetime import timedelta
 from typing import AsyncIterable
 
-from streamstore import S2
-from streamstore.schemas import Record
-from streamstore.utils import append_inputs_gen
+from s2_sdk import S2, Batching, Record, append_inputs
 
 ACCESS_TOKEN = os.getenv("S2_ACCESS_TOKEN")
 MY_BASIN = os.getenv("MY_BASIN")
@@ -23,17 +21,20 @@ async def records_gen() -> AsyncIterable[Record]:
 
 
 async def producer():
-    async with S2(access_token=ACCESS_TOKEN) as s2:
+    async with S2(ACCESS_TOKEN) as s2:
         stream = s2[MY_BASIN][MY_STREAM]
-        async for output in stream.append_session(
-            append_inputs_gen(
+        async with stream.append_session() as session:
+            async for batch in append_inputs(
                 records=records_gen(),
-                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")
+                batching=Batching(
+                    max_records=10,
+                    linger=timedelta(milliseconds=5),
+                ),
+            ):
+                ticket = await session.submit(batch)
+                ack = await ticket
+                num_appended_records = ack.end.seq_num - ack.start.seq_num
+                print(f"appended {num_appended_records} records")
 
 
 if __name__ == "__main__":

examples/producer.py

@@ -0,0 +1,24 @@
+import asyncio
+import os
+import random
+
+from s2_sdk import S2, Record
+
+ACCESS_TOKEN = os.getenv("S2_ACCESS_TOKEN")
+MY_BASIN = os.getenv("MY_BASIN")
+MY_STREAM = os.getenv("MY_STREAM")
+
+
+async def main():
+    async with S2(ACCESS_TOKEN) as s2:
+        stream = s2[MY_BASIN][MY_STREAM]
+        async with stream.producer() as producer:
+            for i in range(100):
+                body_size = random.randint(1, 1024)
+                ticket = await producer.submit(Record(body=os.urandom(body_size)))
+                ack = await ticket
+                print(f"seq_num={ack.seq_num}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/read_session.py

@@ -1,16 +1,15 @@
 import asyncio
 import os
 
-from streamstore import S2
-from streamstore.schemas import SeqNum
+from s2_sdk import S2, SeqNum
 
 ACCESS_TOKEN = os.getenv("S2_ACCESS_TOKEN")
 MY_BASIN = os.getenv("MY_BASIN")
 MY_STREAM = os.getenv("MY_STREAM")
 
 
 async def consumer():
-    async with S2(access_token=ACCESS_TOKEN) as s2:
+    async with S2(ACCESS_TOKEN) as s2:
         stream = s2[MY_BASIN][MY_STREAM]
         tail = await stream.check_tail()
         print(f"reading from tail: {tail}")