feat(py-client): Implement "many" api for batch requests

clients/python/README.md

@@ -155,6 +155,121 @@ existing_parts = resumed.list_parts()
 key = resumed.complete(new_parts + existing_parts)
 ```
 
+### Many API (batch operations)
+
+`session.many()` executes a number of operations with as few requests as
+possible. Those within the batch protocol's per-part limit of 1 MB are grouped
+into requests to Objectstore's batch endpoint which cuts network overhead
+considerably. Inserts too large for that, or of unknown size, are sent as
+individual requests instead.
+
+Pass any iterable of `Get`, `Put`, `Delete`, and `Head` operations, which live in
+the `many` module. Results come back as `GetResult`, `PutResult`, `DeleteResult`,
+and `HeadResult` objects, each carrying the object's `key` and an `error` that is
+`None` when the operation succeeded:
+
+```python
+from objectstore_client import Client, Usecase, many
+
+client = Client("http://localhost:8888")
+session = client.session(Usecase("attachments"), org=42, project=1337)
+
+results = session.many(
+    [
+        many.Put(b"file1 contents", key="file1"),
+        many.Put(b"file2 contents", key="file2"),
+        many.Get("file3"),
+        many.Delete("file4"),
+        many.Head("file5"),
+    ]
+)
+
+for result in results:
+    if result.error is not None:
+        ...  # this operation failed
+    elif isinstance(result, many.GetResult):
+        # `response` is None if the object does not exist.
+        payload = result.response.payload if result.response else None
+```
+
+`session.many()` returns an `OperationResults` object which is a lazy iterator.
+As the iterator is consumed, it assembles batch requests and sends them to
+Objectstore. As responses come in, the operation results are yielded. Abandoning
+the iterator without fully consuming it will cancel whatever has not been
+dispatched yet.
+
+If successful results don't need to be processed or inspected, callers can call
+`raise_for_failures()` to drain the results and raise an `ExceptionGroup` with
+all per-operation errors, or `failures()` which returns the failed results as a
+list:
+
+```python
+session.many([many.Delete("file1"), many.Delete("file2")]).raise_for_failures()
+
+for failure in session.many([many.Delete("file3")]).failures():
+    print(failure.key, failure.error)
+```
+
+#### Concurrency
+
+`concurrency` caps how many requests are in flight, and defaults to `1`. When
+`concurrency` is set to `1` and no `executor` is provided, requests are run
+serially on the caller thread. Otherwise, requests will run on a passed-in
+executor or, if one isn't provided, a thread pool created by `session.many()`.
+Note that, while passing `concurrency=3` (or some other value) is sufficient to
+enable concurrent requests, each request opens and closes new connections;
+consider using the `connection_kwargs` client parameter to enable proper
+connection pooling in `urllib3` as shown below.
+
+```python
+from concurrent.futures import ThreadPoolExecutor
+
+client = Client("http://localhost:8888", connection_kwargs={"maxsize": 8})
+session = client.session(Usecase("attachments"), org=42, project=1337)
+
+with ThreadPoolExecutor(max_workers=8) as executor:
+    for result in session.many(operations, concurrency=8, executor=executor):
+        ...
+```
+
+Results are yielded as responses are received, and the order isn't necessarily
+the same order that operations were given in. Each result carries an `index`
+field that corresponds to the index of the `Get` / `Put` / `Delete` / `Head`
+operation in the operation iterable passed into `session.many()`. This `index`
+allows a keyless `Put` operation to be linked with its result to learn the key
+that was assigned.
+
+```python
+uploads = [many.Put(b"first"), many.Put(b"second")]
+
+for result in session.many(uploads):
+    print(f"{uploads[result.index].contents!r} was stored as {result.key}")
+```
+
+An `ErrorResult` carries `index=None` when the response part it came from could
+not be attributed to any operation at all.
+
+Within a single batch, the Objectstore server processes individual operations
+concurrently and each operation's relative order is undefined. To minimize the
+likelihood of racing operations on the same key, the client will separate
+same-key operations into different batches if at least one of the operations is
+a write or delete. However, with a `concurrency` value larger than `1`, it is
+still possible for same-key operations to race if they are part of separate
+batches that are being processed concurrently. If this is a big concern, pass
+`concurrency=1` to ensure same-key operations never race.
+
+#### Metrics
+
+When a metrics backend is configured, `session.many()` emits some metrics:
+- `storage.batch.latency`: a timer recording a batch request's execution time,
+  tagged with a (bucketed) number of operations included in the batch
+- `stoarge.batch.operations`: a simple counter of individual operations, tagged
+  with each operation's kind (i.e. `PUT`/`GET`/`DELETE`).
+
+An operation that doesn't qualify for batching will be sent through the
+`session`'s regular single-operation API for that operation and will emit
+single-object metrics on that path rather than batch metrics here.
+
 ### Authentication
 
 If your Objectstore instance enforces authorization, you must configure authentication

clients/python/docs/objectstore_client.rst

@@ -33,6 +33,14 @@ objectstore\_client.errors module
    :show-inheritance:
    :undoc-members:
 
+objectstore\_client.many module
+-------------------------------
+
+.. automodule:: objectstore_client.many
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
 objectstore\_client.metadata module
 -----------------------------------
 

clients/python/src/objectstore_client/__init__.py

@@ -1,3 +1,4 @@
+from objectstore_client import many
 from objectstore_client.auth import Permission, SecretKey, TokenGenerator, TokenProvider
 from objectstore_client.client import (
     Client,
@@ -22,6 +23,7 @@
     "Session",
     "GetResponse",
     "RequestError",
+    "many",
     "Compression",
     "ExpirationPolicy",
     "Metadata",

clients/python/src/objectstore_client/client.py

@@ -2,11 +2,11 @@
 
 import math
 import warnings
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime, timedelta
 from io import BytesIO
-from typing import IO, Any, Literal, NamedTuple, cast
+from typing import IO, TYPE_CHECKING, Any, Literal, NamedTuple, cast
 from urllib.parse import urlparse
 
 import sentry_sdk
@@ -39,6 +39,12 @@
 PARAM_AUTH = "os_auth"
 
 
+if TYPE_CHECKING:
+    from concurrent.futures import Executor
+
+    from objectstore_client.many import Operation, OperationResults
+
+
 class GetResponse(NamedTuple):
     metadata: Metadata
     payload: IO[bytes]
@@ -288,6 +294,73 @@ def _make_url(self, key: str | None, full: bool = False) -> str:
             return f"{self._base_url()}{path}"
         return path
 
+    def _make_batch_url(self) -> str:
+        relative_path = f"/v1/objects:batch/{self._usecase.name}/{self._scope}/"
+        return self._base_path.rstrip("/") + relative_path
+
+    def many(
+        self,
+        operations: Iterable[Operation],
+        *,
+        concurrency: int | None = None,
+        executor: Executor | None = None,
+    ) -> OperationResults:
+        """
+        Executes multiple operations, batching them where possible.
+
+        Operations that satisfy the batch protocol's per-part size limit of 1MB
+        are grouped into batch requests to reduce network overhead. Inserts with
+        larger sizes (or unknown sizes) are sent as individual requests instead.
+
+        Args:
+            operations: The operations to execute, as
+                :class:`~objectstore_client.many.Get`,
+                :class:`~objectstore_client.many.Put`,
+                :class:`~objectstore_client.many.Delete`, and
+                :class:`~objectstore_client.many.Head` instances. Any iterable
+                works.
+            concurrency: The maximum number of requests in flight. Defaults to
+                ``1``, which runs everything sequentially on the calling thread,
+                without a thread pool. Raising it is sufficient to send requests
+                concurrently, but each request opens/closes its own connection
+                without additional configuration on the :class:`Client` (a
+                ``maxsize`` key in the ``connection_kwargs`` dict to control
+                ``urllib3`` connection pool size).
+            executor: An executor to run requests on, instead of a thread pool
+                owned by this call. ``concurrency`` still caps how much work is
+                submitted to it at a time, and passing an executor overrides the
+                sequential behavior of ``concurrency=1``.
+
+        Returns:
+            An :class:`~objectstore_client.many.OperationResults` iterator over
+            the results. Results are yielded as responses come in, in no
+            particular order; each carries the ``index`` of the operation it
+            belongs to, which is the only handle on a keyless
+            :class:`~objectstore_client.many.Put`. This iterator is lazy, and if
+            it's abandoned without being fully consumed then operations that
+            haven't been dispatched are cancelled.
+
+        Raises:
+            ValueError: If ``concurrency`` is less than ``1``.
+
+        Example::
+
+            from objectstore_client import many
+
+            results = session.many([many.Put(b"hello", key="k1"), many.Get("k2")])
+            for result in results:
+                if result.error is not None:
+                    ...  # this operation failed
+                elif isinstance(result, many.GetResult):
+                    ...  # `result.response` is None if the object does not exist
+        """
+        # Imported lazily to avoid a circular import at module load time.
+        from objectstore_client.many import execute_many
+
+        return execute_many(
+            self, operations, concurrency=concurrency, executor=executor
+        )
+
     def _make_multipart_url(
         self,
         action: str | None,