cosmodol

Access Azure Cosmos DB (NoSQL/Core API) through a Mapping interface.

cosmodol exposes Azure Cosmos DB as dol-style Mapping / MutableMapping interfaces, layered over the official azure-cosmos SDK.

Quick start:

from cosmodol import cosmos_store

store = cosmos_store(
    connection_string="AccountEndpoint=https://localhost:8081/;AccountKey=...",
    database="mydb",
    container="mycontainer",
    partition_key_value="tenant-X",
)

store["k1"] = {"name": "Alice", "age": 30}
store["k1"]              # → {"id": "k1", "<pk>": "tenant-X", "name": "Alice", "age": 30}
"k1" in store            # → True
del store["k1"]

See misc/docs/architecture.md for the layered design.

If your Cosmos account was provisioned with the MongoDB API, use pymongo + mongodol directly — this package only targets the NoSQL/Core API.

exception cosmodol.ContainerNotEmptyError

Raised on del db_store[name] when the container has items. See misc/docs/design_decisions.md §8.

exception cosmodol.ContainerNotFoundError

Raised when a Cosmos container does not exist.

class cosmodol.CosmosAccount(connection: Any = None, *, database_factory: Callable | None = None)

Mapping of database name → CosmosDatabase.

__setitem__ is disabled — use add_database(name, throughput=...). __delitem__ refuses non-empty databases; use self.delete(name, force=True).

delete(k: str, *, force: bool = False) → None

Delete a database; if force=True, cascade-delete its containers first.

class cosmodol.CosmosConnection(credential: str | dict | Any | None = None, connection_string: str | None = None, endpoint: str | None = None, key: str | None = None, consistency_level: str | None = None, client_kwargs: dict = <factory>)

Holds a resolved credential and a lazy CosmosClient.

This is the dependency-injection seam for the package. Tests construct one pointing at the emulator without touching any store class.

Parameters:

• credential – explicit credential object or account-key string.

• connection_string – full AccountEndpoint=...;AccountKey=... string.

• endpoint – Cosmos endpoint URL (with or without credential).

• key – account master key.

• consistency_level – optional override; defaults to inheriting the account default.

• client_kwargs – extra kwargs forwarded to CosmosClient.

property client: CosmosClient

The lazily-constructed CosmosClient. Cached for the connection’s lifetime.

classmethod from_anything(source) → CosmosConnection

Convenience: build a CosmosConnection from a thing-or-spec.

Accepts:

• CosmosConnection (returned as-is)

• CosmosClient (wrapped without further resolution)

• str (connection string)

• dict (passed as kwargs)

• None (defer to env / AAD)

class cosmodol.CosmosDatabase(database: DatabaseProxy | str, *, connection: Any = None, store_factory: Callable | None = None)

Mapping of container name → ContainerProxy (or store-wrapped form).

__setitem__ is disabled — use add_container(name, partition_key_path=..., ...). __delitem__ refuses non-empty containers; use self.delete(name, force=True).

Parameters:

• database – Either an existing DatabaseProxy, a database name string (resolved via connection), or None (use connection.database_name).

• connection – CosmosConnection (or anything from_anything accepts).

• store_factory – Optional callable (ContainerProxy) -> Mapping used to wrap the value returned by __getitem__. Defaults to passing the ContainerProxy through.

add_container(name: str, *, partition_key_path: str = '/id', throughput: int | None = None, indexing_policy: dict | None = None, default_ttl: int | None = None, unique_key_policy: dict | None = None, conflict_resolution_policy: dict | None = None, **extra)

Create a container.

Parameters:

• name – Container id.

• partition_key_path – Partition-key path (e.g. "/id", "/tenantId"). Defaults to "/id" per misc/docs/design_decisions.md §3.

• throughput – Provisioned RU/s (e.g. 400). None uses the database’s shared throughput if any, otherwise the account default.

• indexing_policy – Standard Cosmos container kwargs.

• default_ttl – Standard Cosmos container kwargs.

• unique_key_policy – Standard Cosmos container kwargs.

• conflict_resolution_policy – Standard Cosmos container kwargs.

• **extra – Forwarded to DatabaseProxy.create_container.

delete(k: str, *, force: bool = False) → None

Delete a container; if force=True, cascade-delete its items first.

class cosmodol.CosmosItems(container: ContainerProxy | dict | tuple, *, partition_key_value: Any, partition_key_path: str | None = None, connection: Any = None, inject_id: bool = True, inject_partition_key: bool = True, strict_keys: bool = True, strip_system_fields: bool = True, record_ru: Callable[[str, float], None] | None = None)

MutableMapping[str, dict] over one fixed partition of a Cosmos container.

Keys are item id strings; values are JSON-dict items. The store auto-injects id and the partition-key property on writes.

Parameters:

• container – An already-built ContainerProxy or a (database, container) tuple / {"database": ..., "container": ...} dict to resolve via connection.

• partition_key_value – The single partition-key value this store is scoped to.

• partition_key_path – Path of the partition-key property in items (e.g. "/_pk", "/id"). Read from the container if absent.

• connection – CosmosConnection (or anything from_anything accepts). Ignored if container is a ContainerProxy.

• inject_id – If True, auto-inject "id" into bodies on writes.

• inject_partition_key – If True, auto-inject the partition-key property into bodies.

• strict_keys – Validate id chars + length on writes.

• strip_system_fields – Strip _etag/_ts/_rid/_self/_attachments from returned items.

• record_ru – Optional callback (op_name, ru) -> None invoked after each metal-layer op. Useful for Prometheus / logging.

batch(operations: list[tuple]) → list[dict]

Transactional batch within this partition. See base.batch.

query(sql: str, *, parameters: list[dict] | None = None) → Iterator[dict]

Run a SQL query scoped to this partition. Yields raw item dicts (no stripping).

replace(k: str, v: dict, *, etag: str | None = None) → dict

Full replace with optional ETag-conditional write.

class cosmodol.CosmosPartitionedItems(container: ContainerProxy | dict | tuple, *, partition_key_path: str | None = None, connection: Any = None, inject_id: bool = True, inject_partition_key: bool = True, strict_keys: bool = True, strip_system_fields: bool = True, record_ru: Callable[[str, float], None] | None = None, len_via_query: bool = False, silent_full_scan: bool = False)

MutableMapping[tuple[str, str], dict] over all partitions of a container.

Keys are (partition_key_value, id) tuples. Use CosmosItems instead if you have a fixed partition.

Iteration is cross-partition; the first call emits a UserWarning (silence with silent_full_scan=True). __len__ is not implemented by default; opt-in via len_via_query=True. See misc/docs/design_decisions.md §§4, 6.

Parameters:

• container – As in CosmosItems.

• partition_key_path – Path of the partition-key property in items. Read from the container if absent.

• connection – As in CosmosItems.

• inject_id – As in CosmosItems.

• inject_partition_key – As in CosmosItems.

• strict_keys – As in CosmosItems.

• strip_system_fields – As in CosmosItems.

• record_ru – As in CosmosItems.

• len_via_query – If True, __len__ runs a (cross-partition) COUNT query.

• silent_full_scan – If True, __iter__ does not emit the cross-partition warning.

partition(pk_value) → CosmosItems

Narrow to a single partition; zero round-trips. Returns CosmosItems.

query(sql: str, *, parameters: list[dict] | None = None, partition_key: Any = None, cross_partition: bool = False) → Iterator[dict]

Pass-through to base.query.

exception cosmodol.CosmosThrottleError

Wraps CosmosHttpResponseError with HTTP 429 (RU exhaustion).

exception cosmodol.DatabaseNotEmptyError

Raised on del account_store[name] when the database has containers. See misc/docs/design_decisions.md §8.

exception cosmodol.DatabaseNotFoundError

Raised when a Cosmos database does not exist.

exception cosmodol.ItemAlreadyExistsError

Raised on strict-create attempts when the item already exists.

exception cosmodol.ItemNotFoundError

Raised when an item does not exist for a (partition_key, id).

exception cosmodol.KeyMismatchError

Raised when a written body’s id (or partition-key value) disagrees with the inferred value (from the dict key). See misc/docs/design_decisions.md §9.

class cosmodol.ResponseHeaders(request_charge: float | None, etag: str | None)

Subset of Cosmos response headers we surface for observability.

request_charge

RU consumed by the operation. None when the emulator (which does not populate this header) was the backend.

Type:

float | None

etag

ETag of the (created / read / replaced) item, when applicable.

Type:

str | None

etag: str | None

Alias for field number 1

request_charge: float | None

Alias for field number 0

cosmodol.batch(container: ContainerProxy, operations: list[tuple], partition_key: Any) → list[dict]

Transactional batch within one logical partition.

Parameters:

• operations – List of (op_name, args_tuple, kwargs_dict) triples. Cosmos op names: "create", "upsert", "replace", "patch", "read", "delete". ≤ 100 ops, ≤ 1.2 MB total.

• partition_key – All ops must share this partition-key value.

Returns:

List of per-op result dicts as returned by the SDK.

cosmodol.cosmos_store(*, database: str, container: str, connection: CosmosConnection | Any | None = None, credential: Any = None, connection_string: str | None = None, endpoint: str | None = None, key: str | None = None, partition_key_value: Any = None, partition_key_path: str | None = None, value_codec: Callable | None = None, strip_system_fields: bool = True)

Build a ready-to-use Cosmos store (CosmosItems or CosmosPartitionedItems).

The store flavor is chosen based on which partition-key kwarg is passed:

• partition_key_value=... → CosmosItems (keys = id strings, fixed partition).

• partition_key_path=... → CosmosPartitionedItems (keys = (pk, id) tuples).

• Neither given → raises ValueError.

Parameters:

• database – Database name.

• container – Container name.

• connection – A CosmosConnection (or anything from_anything accepts) to reuse a client. Mutually exclusive with explicit credential kwargs.

• credential – Forwarded to CosmosConnection when connection is None.

• connection_string – Forwarded to CosmosConnection when connection is None.

• endpoint – Forwarded to CosmosConnection when connection is None.

• key – Forwarded to CosmosConnection when connection is None.

• partition_key_value – Fixed partition key value → returns CosmosItems.

• partition_key_path – Partition key property path → returns CosmosPartitionedItems.

• value_codec – A decorator that, given a class, returns a wrapped class. Typically a partially-applied dol.wrap_kvs(...).

• strip_system_fields – Strip Cosmos system fields from returned items.

Returns:

A CosmosItems or CosmosPartitionedItems instance, possibly codec-wrapped.

cosmodol.point_contains(container: ContainerProxy, id: str, partition_key: Any) → tuple[bool, ResponseHeaders]

Existence check via point read. True/False. Never raises KeyError.

Cheaper than a SELECT VALUE COUNT(1) query — ~1 RU vs ≥ 2.3 RU.

cosmodol.point_delete(container: ContainerProxy, id: str, partition_key: Any, *, etag: str | None = None) → ResponseHeaders

Point delete. Raises ItemNotFoundError on missing.

cosmodol.point_get(container: ContainerProxy, id: str, partition_key: Any) → tuple[dict, ResponseHeaders]

Point read of one item. ~1 RU/KB. Raises ItemNotFoundError on missing.

cosmodol.point_replace(container: ContainerProxy, id: str, body: dict, partition_key: Any, *, etag: str | None = None) → tuple[dict, ResponseHeaders]

Full replace of an item. With etag, performs an If-Match conditional write.

cosmodol.point_upsert(container: ContainerProxy, body: dict, *, etag: str | None = None) → tuple[dict, ResponseHeaders]

Insert-or-replace an item. id and the partition-key value are extracted from body.

Note: etag parameter accepted for symmetry but Cosmos does not honor it on upsert_item; use point_replace for ETag-conditional writes.

cosmodol.query(container: ContainerProxy, sql: str, *, parameters: list[dict] | None = None, partition_key: Any = None, cross_partition: bool = False, max_item_count: int | None = None) → Iterator[dict]

Run a SQL query. Yields dicts.

Either pass partition_key= (cheap, single-partition) or cross_partition=True (RU scales with data). One of the two is required by Cosmos.

cosmodol.resolve_credential(*, credential: str | dict | Any | None = None, connection_string: str | None = None, endpoint: str | None = None, key: str | None = None) → dict

Resolve a credential into a normalized form that can build a CosmosClient.

Cascade (first hit wins):

1. Explicit credential= (with endpoint= for the URL)

2. Explicit connection_string=

3. Explicit endpoint= + key= (or just endpoint= + AAD)

4. Env var AZURE_COSMOS_CONNECTION_STRING

5. Env vars AZURE_COSMOS_ENDPOINT + AZURE_COSMOS_KEY

6. Env var AZURE_COSMOS_ENDPOINT alone + DefaultAzureCredential

Returns:

{"url": "...", "credential": <obj>}

Return type:

A dict with

Raises:

ValueError – if no source resolves.

cosmodol.strip_system_fields(item: dict) → dict

Return item with Cosmos system fields removed.

cosmodol.translate_cosmos_errors(*, key_arg: int | str = 0, not_found_cls: type[KeyError] = <class 'cosmodol.errors.ItemNotFoundError'>, exists_cls: type[KeyError] = <class 'cosmodol.errors.ItemAlreadyExistsError'>) → Callable

Decorator: translate Cosmos SDK exceptions into KeyError subclasses.

Auth errors and any other Cosmos errors propagate untouched. HTTP 429 throttling is wrapped in CosmosThrottleError so callers can distinguish it. See misc/docs/design_decisions.md §11.

Parameters:

• key_arg – Position (int) or name (str) of the key argument in the wrapped method’s signature. Used to populate KeyError(key). Default 0; for methods the user-facing key is typically at index 1 (self at 0).

• not_found_cls – Exception class to raise on CosmosResourceNotFoundError.

• exists_cls – Exception class to raise on CosmosResourceExistsError.

cosmodol.validate_cosmos_id(k) → None

Raise ValueError if k is not a valid Cosmos item id.

Cosmos may accept invalid ids on write but the item then becomes unreachable from the SDK. We fail loudly at write time. See misc/docs/cosmos_db_reference.md §”id rules”.