feat: add asyncio client in orbita.aio

README.md

@@ -1,8 +1,9 @@
 # Orbita Python
 
-`orbita-python` is a synchronous Python client for Orbita. It connects to one
-load-balanced endpoint and leaves partition routing to the server, which is the
-same complete non-transactional posture supported by Orbita's generated stubs.
+`orbita-python` is a Python client for Orbita, with synchronous and asyncio
+APIs. It connects to one load-balanced endpoint and leaves partition routing to
+the server, which is the same complete non-transactional posture supported by
+Orbita's generated stubs.
 
 The package is pinned to the wire contract at
 `orbita 0.2.0-dev@cb11cf03575e2c6e370fb83500038de45bd55e69`. Generated code
@@ -36,6 +37,27 @@ with orbita.Client(
         print(f"already exists at version {created.current_version}")
 ```
 
+The asyncio API lives in `orbita.aio` and mirrors the synchronous one
+method-for-method, sharing the same result and error types. The one difference
+is that connection setup is explicit, because `__init__` cannot await: use
+`async with`, which connects on entry, or call `await client.connect()`.
+
+```python
+from orbita import aio
+
+async with aio.Client(
+    "http://127.0.0.1:7100",
+    credential="tenant-secret",
+) as client:
+    catalog = client.keyspace("default")
+    entry = await catalog.get_entry(b"catalog/current")
+    async for page in catalog.pages(b"catalog/", include_values=True):
+        ...
+```
+
+`orbita.aio.AdminClient` is the asynchronous counterpart of
+`orbita.admin.AdminClient`.
+
 Use an explicit `http://` endpoint for plaintext h2c. Use `https://` for TLS;
 pass `tls_credentials=grpc.ssl_channel_credentials(...)` when the endpoint
 needs custom trust roots or a client certificate.

pyproject.toml

@@ -39,6 +39,7 @@ dev = [
     "grpcio-tools==1.75.1",
     "mypy>=1.14,<2",
     "pytest>=8.3,<9",
+    "pytest-asyncio>=0.25,<2",
     "pytest-cov>=6,<8",
     "ruff>=0.9,<1",
     "types-protobuf>=6.31.1,<7",
@@ -51,6 +52,7 @@ packages = ["src/orbita"]
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 addopts = "-q -ra"
+asyncio_mode = "auto"
 markers = ["integration: requires ORBITA_ENDPOINT and ORBITA_ROOT_CREDENTIAL"]
 
 [tool.ruff]

src/orbita/__init__.py

@@ -1,4 +1,8 @@
-"""Synchronous Python client for Orbita."""
+"""Python client for Orbita.
+
+The synchronous API is exported here; the asyncio API lives in ``orbita.aio``
+and is not imported eagerly so synchronous users never touch ``grpc.aio``.
+"""
 
 from orbita._transport import AmbiguousMutationError, CredentialError, RetryPolicy
 from orbita.client import (

src/orbita/_transport.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
 from dataclasses import dataclass
-from typing import Any, Callable, Optional, Sequence, Tuple, TypeVar, cast
+from typing import Any, Awaitable, Callable, Optional, Sequence, Tuple, TypeVar, cast
 from urllib.parse import SplitResult, urlsplit
 
 import grpc
+import grpc.aio
 
 T = TypeVar("T")
 Metadata = Sequence[Tuple[str, str]]
@@ -102,6 +104,21 @@ def open_channel(self, max_message_bytes: int = 0) -> grpc.Channel:
         credentials = self.tls_credentials or grpc.ssl_channel_credentials()
         return grpc.secure_channel(target, credentials, options=options)
 
+    def open_aio_channel(self, max_message_bytes: int = 0) -> grpc.aio.Channel:
+        options = list(self.channel_options)
+        if max_message_bytes:
+            options.extend(
+                (
+                    ("grpc.max_receive_message_length", max_message_bytes),
+                    ("grpc.max_send_message_length", max_message_bytes),
+                )
+            )
+        target = self.endpoint.netloc
+        if self.endpoint.scheme == "http":
+            return grpc.aio.insecure_channel(target, options=options)
+        credentials = self.tls_credentials or grpc.ssl_channel_credentials()
+        return grpc.aio.secure_channel(target, credentials, options=options)
+
     def metadata(self) -> Metadata:
         if self.token_source is None:
             return ()
@@ -147,6 +164,37 @@ def retry_read(
     raise RuntimeError("orbita: retry loop exhausted")
 
 
+async def retry_read_aio(
+    call: Callable[..., Awaitable[T]],
+    request: Any,
+    config: TransportConfig,
+    timeout: Optional[float],
+) -> T:
+    deadline = None if timeout is None else time.monotonic() + _validate_timeout(timeout)
+    backoff = config.retry_policy.initial_backoff
+    for attempt in range(1, config.retry_policy.max_attempts + 1):
+        metadata = config.metadata()
+        remaining = _remaining(deadline)
+        if remaining == 0:
+            raise TimeoutError("orbita: operation timed out")
+        try:
+            return await call(request, timeout=remaining, metadata=metadata)
+        except grpc.RpcError as error:
+            if error.code() != grpc.StatusCode.UNAVAILABLE:
+                raise
+            if attempt == config.retry_policy.max_attempts:
+                raise
+        sleep_for = backoff
+        if deadline is not None:
+            sleep_for = min(sleep_for, max(0.0, deadline - time.monotonic()))
+        if sleep_for > 0:
+            await asyncio.sleep(sleep_for)
+        if deadline is not None and time.monotonic() >= deadline:
+            raise TimeoutError("orbita: operation timed out")
+        backoff = min(backoff * 2, config.retry_policy.max_backoff)
+    raise RuntimeError("orbita: retry loop exhausted")
+
+
 def mutation_error(operation: str, error: grpc.RpcError, *, aborted: bool = False) -> Exception:
     ambiguous = {
         grpc.StatusCode.UNAVAILABLE,