RociaDB Python SDK

Typed, async Python client for the RociaDB document, graph, file, and tenant gRPC services. It covers all 22 RPCs exposed by the packaged protobuf and adds pagination helpers, JSON encoding/decoding, bounded batch operations, OAuth2 client-credentials auth with automatic background token refresh, and ergonomic file streaming, so callers work with plain Python values instead of hand-building protobuf messages.

Contents

• What the SDK Manages
• Requirements and Installation
• Connecting
• Authentication
• Pagination
• Documents
• Graph
• Files
• Tenants
• Idempotence and Retries
• Error Handling
• API Coverage
• Parity with the Other SDKs
• Development and Testing

What the SDK Manages

RociaDB organizes data around four identifiers:

• A tenant_id segments data by customer or workspace. Every RPC except list_tenants requires one.
• A document collection groups JSON documents addressed by document_id.
• A graph groups nodes and directed edges. Node IDs conventionally use label:id, for example product:sku-123.
• A file bucket groups binary objects addressed by file_id.

Document, node, and edge payloads are plain JSON-serializable Python values (dict, list, str, int, float, bool, None). Read methods that decode a payload accept an optional decoder keyword argument to turn the raw JSON value into a caller-supplied type; the SDK itself performs no runtime schema validation beyond "is this valid JSON".

tenant_id is not a security boundary. It is not derived from the caller's identity, so any authenticated client can address any tenant_id — it exists to segment data for application-level purposes, not to isolate customers from each other at the protocol level. Per-user or per-customer authorization is the calling application's responsibility; see Authentication for what the server does check.

Requirements and Installation

• Python 3.10 through 3.14
• A reachable RociaDB gRPC endpoint
• OAuth2 client credentials, unless authentication is explicitly disabled

pip install rociadb-sdk

or, with uv:

uv add rociadb-sdk

The only runtime dependencies are grpcio and protobuf — the OAuth2 token exchange is done with urllib.request from the standard library, not a third-party HTTP client. The SDK is asynchronous, built on grpc.aio, and ships full inline type annotations (a py.typed marker) that pass mypy --strict in a consumer's own codebase.

Connecting

Use the builder when you want fluent configuration:

from rociadb_sdk import RociaDbBuilder

client = await (
    RociaDbBuilder()
    .host("https://db.example.com:443")
    .auth_client_credentials(
        "https://auth.example.com/oauth/token",
        "client-id",
        "client-secret",
    )
    .connect_timeout(15.0)
    .build()
)

http:// creates an insecure gRPC channel and https:// enables TLS. RociaDB servers do not terminate TLS themselves — they listen in clear text — so https:// is only meaningful when it points at a reverse proxy sitting in front of the server; that proxy is the normal production endpoint, typically on port 443 (a bare https://host with no explicit port defaults to it, the same way http://host defaults to port 80). Point http://host:50051 (the default gRPC port) directly at a RociaDB process for a local or unproxied connection instead.

When .auth_client_credentials(...) is not called and auth is not disabled, each of the token URL, client ID, and client secret falls back independently to an environment variable:

AUTH_TOKEN_URL
AUTH_CLIENT_ID
AUTH_CLIENT_SECRET

For a controlled local environment, authentication can be disabled explicitly:

client = await RociaDbBuilder().host("http://127.0.0.1:50051").disable_auth().build()

A direct, non-fluent constructor covers the same options in one call:

from rociadb_sdk import RociaDbClient

client = await RociaDbClient.connect(
    "https://db.example.com:443",
    auth_token_url="https://auth.example.com/oauth/token",
    auth_client_id="client-id",
    auth_client_secret="client-secret",
    connect_timeout=15.0,
)

Omitting host defaults to "http://127.0.0.1:50051"; omitting any of the three auth_* keyword arguments falls back to its environment variable exactly as the builder does. RociaDbClient also supports the async context-manager protocol, which calls close() automatically:

async with await RociaDbClient.connect(disable_auth=True) as client:
    tenants = await client.list_tenants()

Create one client per upstream configuration and reuse it — a single RociaDbClient instance is already safe to call concurrently from many asyncio tasks, with no locking or cloning step needed. Call close() during graceful shutdown; the client must not be reused afterward.

.connect_timeout(seconds) sets the deadline applied while every service connects; call it on the builder before .build(), or pass connect_timeout directly to RociaDbClient.connect(...). When neither is supplied, the client falls back to its own default of 10.0 seconds (DEFAULT_CONNECT_TIMEOUT), so a slow or unreachable DNS/TCP target fails after a bounded wait instead of hanging forever. A non-positive value raises RociaDbValidationError immediately when .connect_timeout(...) is called, before any connection attempt is made.

.host(...) must resolve to a bare hostname and port — no path component beyond an absent one or a lone /. A mistyped host with a leftover path (http://127.0.0.1:50051/v1, pasted from somewhere else) is rejected with RociaDbConnectionError before any connection attempt, rather than silently dialing the host and dropping the path.

Authentication

Every call carries a bearer token as gRPC metadata:

authorization: Bearer <jwt>

The SDK obtains it from the configured token URL with a standard OAuth2 client-credentials request (POST, application/x-www-form-urlencoded, grant_type=client_credentials&client_id=...&client_secret=...) and expects a response shaped like:

{ "access_token": "...", "token_type": "Bearer", "expires_in": 600 }

Tokens issued by RociaDB's identity provider are valid for 600 seconds (10 minutes), fixed server-side. The SDK does not hardcode that number — it reads expires_in from each response — but in practice it is always 600.

A background task refreshes the cached token on its own schedule for as long as the client is open, derived from the last reported token lifetime: max(expires_in * 2 // 3, 5.0) seconds — about 400 seconds for the IdP's fixed 600-second lifetime, leaving margin so the token never actually expires between two refreshes. As long as the client stays open, you never need to schedule or poll for a refresh yourself.

UNAUTHENTICATED vs PERMISSION_DENIED

These two statuses look similar but call for opposite handling:

• UNAUTHENTICATED — the token is missing, expired, malformed, or signed by a different issuer. Treat this as a renewal signal: refresh the token and retry.
• PERMISSION_DENIED — the token is valid but its scope does not cover the operation. Retrying does not help, even after a refresh, because a fresh token carries the same scope. Two causes exist:
  • a read-only scoped client called one of the 7 write RPCs: put_document/create_document, delete_document, put_node/put_nodes, add_edge/add_edges, delete_edge, upload_file/upload_file_chunked/upload_file_raw, or delete_file. A read-only token is not otherwise crippled — all 15 read RPCs remain available.
  • an admin-scoped token was presented — the credentials used to manage rocia-idp service accounts, not to read or write data. It is rejected on all 22 RPCs, reads included. If a read you expect to work returns PERMISSION_DENIED, check which client_id produced the token: the data-plane account and the administration account are different credentials.

from rociadb_sdk import RociaDbStatusError

try:
    await client.get_document("tenant-1", "products", "sku-123")
except RociaDbStatusError as error:
    if error.is_unauthenticated():
        await client.refresh_auth_token()
        await client.get_document("tenant-1", "products", "sku-123")  # retry once
    else:
        raise  # includes PERMISSION_DENIED: retrying will not help

is_unauthenticated() and is_permission_denied() are defined on the base RociaDbError too (always returning False there), so either predicate is safe to call on any caught error without checking the exception type first.

Two ways to recover from UNAUTHENTICATED

client.invalidate_auth_token() is the lazy option: synchronous, returns immediately, and makes no network call itself — it only wakes the background refresh task described above so it refreshes sooner. The cached token keeps being used for any call made before that background refresh actually completes, so calling invalidate_auth_token() immediately before retrying can still send the same, still-invalid token. refresh_auth_token() is the eager counterpart: it awaits the round trip to the identity provider itself and only returns once a fresh token is confirmed and cached, or raises RociaDbAuthError on failure — the right choice immediately before retrying the call that just failed, as shown above.

Reach for invalidate_auth_token() instead when you just want to mark the token stale without blocking on a fresh one right now — a fire-and-forget error handler that is not about to retry immediately, for example. Both are no-ops when the client was built with disable_auth(). Neither ever discards a still-valid cached token just because a refresh attempt failed: a failed background refresh is logged and retried on the next tick rather than clearing the cache.

Auth helpers outside a client

fetch_token and TokenManager are importable from rociadb_sdk.auth for callers who need OAuth2 token handling outside of a RociaDbClient — to reuse the same access token against a different service, for example. They are not re-exported from the rociadb_sdk package root, since they are useful independently of any RociaDbClient instance:

from rociadb_sdk.auth import TokenManager, fetch_token

# One-off token exchange, no caching or refresh:
token = await fetch_token("https://auth.example.com/oauth/token", "client-id", "client-secret")
print(token.access_token, token.token_type, token.expires_in)

# Cached, self-refreshing token manager - the same one RociaDbClient uses internally:
token_manager = TokenManager("https://auth.example.com/oauth/token", "client-id", "client-secret")
header = await token_manager.get_authorization_header()  # "Bearer <token>"

fetch_token performs one token exchange and returns; it does not cache anything. TokenManager(...) fetches nothing until its first get_authorization_header() call (or an explicit await token_manager.refresh_now()); starting its own background refresh loop requires token_manager.spawn_refresh(token_manager.refresh_interval()), which returns a TokenRefreshHandle to await ....aclose() on shutdown.

Pagination

Every listing method — documents, collections, graphs, nodes, neighbors, buckets, files, and tenants — takes keyword-only limit/cursor arguments and returns a Page[T] (items, next_cursor), or a DocumentPage[T] which additionally carries total_count.

• limit must be a positive integer. The SDK rejects 0 and negative values client-side, before any RPC, with RociaDbValidationError. Omit it to use the SDK's own default of 20 (DEFAULT_PAGE_SIZE).

• The server enforces its own ceiling, limits.max_page_size (200 by default, operator-configurable per deployment). A limit above that ceiling is rejected server-side with INVALID_ARGUMENT — the SDK does not hardcode 200, or any other ceiling, client-side.

• An absent next_cursor is the only end-of-list signal. Loop while it is not None:

  from typing import Optional
  
  cursor: Optional[str] = None
  while True:
      page = await client.list_documents("tenant-1", "products", limit=100, cursor=cursor)
      for item in page.items:
          print(item)
      if page.next_cursor is None:
          break
      cursor = page.next_cursor
  

• Do not stop because a page is short, or even empty. find_documents_by_field, list_documents, and query_documents in particular can return fewer items than limit — or none at all — in the middle of a paginated walk, when an index entry briefly survives the document it points to being deleted. A short or empty page with a non-None next_cursor is not the end; only the cursor tells you that.

• An exact-multiple total produces one extra, empty page. If a collection holds exactly limit items, the final full page still carries a cursor — the server has no way to know it just emitted everything — so the next call returns an empty page with next_cursor=None. This is expected, not a bug; the loop above handles it correctly by construction.

• Cursors are opaque: never construct, parse, or persist one across sessions. Their shape differs per RPC and may change.

list_graphs, list_buckets, and list_nodes paginate over graph/bucket names, not the items each one contains — a graph or bucket name appears once no matter how many nodes or files it holds.

Documents

Create, read, and delete

product = {
    "sku": "sku-123",
    "label": "Widget",
    "active": True,
    "price": 19.9,
}

await client.put_document("tenant-1", "products", "sku-123", product)

fetched = await client.get_document("tenant-1", "products", "sku-123")

await client.delete_document("tenant-1", "products", "sku-123")

get_document accepts an optional decoder keyword to turn the raw JSON value into any type you like:

from dataclasses import dataclass


@dataclass
class Product:
    sku: str
    label: str
    active: bool
    price: float


typed = await client.get_document(
    "tenant-1", "products", "sku-123", decoder=lambda raw: Product(**raw)
)

put_document replaces the document completely — there is no partial merge — and recomputes its indexes (exact-match and trigram) in the same transaction. Its JSON payload is capped at the server's limits.max_doc_bytes (2 MiB by default); a larger payload is rejected with INVALID_ARGUMENT before anything is written. delete_document is idempotent: deleting a document that does not exist is not an error (contrast this with delete_edge in Directed edges and neighbors, which is not).

Create a document and its graph reference

create_document first stores the document, then optionally creates a graph node whose payload points back to {"collection": ..., "id": ...}:

await client.create_document(
    "tenant-1",
    "products",
    "sku-123",
    product,
    node_label="product",
    node_graph="catalog",
)
# Creates document products/sku-123 and node product:sku-123 in catalog.

node_label and node_graph must be supplied together — passing only one raises RociaDbValidationError before any network call. This is a composite, non-atomic operation: if the node write fails, the document remains stored and the caller should retry or compensate.

Search, query, and pagination

Use find_documents_by_field for one exact field lookup, list_documents for an unfiltered collection, and query_documents for multiple filters and sorting:

by_sku = await client.find_documents_by_field("tenant-1", "products", "sku", "sku-123")

find_documents_by_field's value must encode to a JSON scalar — a string, number, boolean, or None. An object or array raises INVALID_ARGUMENT server-side; it uses the same exact-match index as query_documents's EQ operator.

from typing import Optional

from rociadb_sdk import (
    DocumentQueryFilter,
    DocumentQueryOperator,
    DocumentQuerySort,
    DocumentSortDirection,
)

cursor: Optional[str] = None
while True:
    page = await client.query_documents(
        "tenant-1",
        "products",
        filters=[
            DocumentQueryFilter(field="active", operator=DocumentQueryOperator.EQ, values=[True]),
            DocumentQueryFilter(
                field="label", operator=DocumentQueryOperator.CONTAINS, values=["Widget"]
            ),
        ],
        sort=[DocumentQuerySort(field="price", direction=DocumentSortDirection.ASC)],
        limit=50,
        cursor=cursor,
    )
    for item in page.items:
        print(item["sku"])
    print("matching documents:", page.total_count)
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

Supported query operators are EQ, IN, and CONTAINS. Filters are sent in order and combined by the server with an implicit AND — there is no OR. CONTAINS is a case-insensitive substring match backed by a trigram index, with two restrictions: a CONTAINS term shorter than 3 characters is not indexable, and a query where no filter is indexable is rejected with INVALID_ARGUMENT rather than served by a full scan — pair a short CONTAINS term with an EQ or IN filter on another field. Cursors are opaque: pass next_cursor back unchanged.

total_count is not uniformly cheap. list_documents's count is read from a counter maintained on every write, so it costs nothing extra. find_documents_by_field's count is an index count over matching entries — not free, but cheaper than a full evaluation. query_documents's count is computed by evaluating the full filtered result set, so it scales with the number of matching documents — prefer list_documents (no filters) when you just need a count, and avoid calling query_documents in a loop purely to read total_count.

Discovering collections

list_collections returns the collections that hold at least one document, each with its document count:

collections = await client.list_collections("tenant-1", limit=50)
for info in collections.items:
    print(info.collection, info.count)

Graph

Nodes and batches

await client.put_node("tenant-1", "catalog", "product:sku-123", product)
node = await client.get_node("tenant-1", "catalog", "product:sku-123")

put_node's payload must encode to a JSON object — not a scalar or an array — and, like put_document, is capped at the server's limits.max_doc_bytes (2 MiB by default).

from rociadb_sdk import NodeInput

await client.put_nodes(
    "tenant-1",
    "catalog",
    [
        NodeInput(node_id="product:sku-124", value={**product, "sku": "sku-124"}),
        NodeInput(node_id="group:featured", value={"title": "Featured"}),
    ],
)

Batch helpers (put_nodes, add_edges, and the neighbor-node helpers below) issue at most CONCURRENT_REQUESTS (10) requests concurrently. They are not atomic: if one item fails, earlier items may already have been stored, and every other already-dispatched call in the batch is cancelled rather than left to run to completion. Retry a failed batch by reusing the same request_id values on each item — see Idempotence and Retries.

Directed edges and neighbors

An edge goes from one node to another. For product:sku-123 -> group:featured, the group is an outgoing neighbor of the product, while the product is an incoming neighbor of the group.

await client.add_edge(
    "tenant-1",
    "catalog",
    "membership-1",
    "product:sku-123",
    "group:featured",
    "belongs_to",
    {"weight": 1},
)

outgoing = await client.neighbors_out(
    "tenant-1", "catalog", "product:sku-123", "belongs_to", limit=25
)

products = await client.get_incoming_neighbor_nodes(
    "tenant-1", "catalog", "group:featured", "belongs_to"
)

await client.delete_edge("tenant-1", "catalog", "membership-1")

add_edge takes the endpoint node IDs as from_id/to_id — from is a reserved word in Python and cannot be used as a parameter name. It fails with NOT_FOUND if either endpoint does not already exist as a node in graph — create both endpoint nodes before the edge — and, like put_document and put_node, its JSON payload is capped at the server's limits.max_doc_bytes (2 MiB by default). delete_edge also fails with NOT_FOUND if the edge itself does not exist; unlike delete_document, deleting an edge is not idempotent.

neighbors_out and neighbors_in return one raw Page[Neighbor], each item carrying node_id and edge_id. get_outgoing_neighbor_nodes and get_incoming_neighbor_nodes follow every page internally and load each neighbor's JSON node payload with the same bounded concurrency as put_nodes (both also accept a decoder keyword, like get_node).

add_edges(tenant_id, graph, edges) batches edge creation from a list of EdgeInput the same way put_nodes batches NodeInput — see Nodes and batches for the concurrency and non-atomicity rules shared by every batch helper.

Discovering graphs and nodes

graphs = await client.list_graphs("tenant-1")
nodes = await client.list_nodes("tenant-1", "catalog", limit=100)

Both return a Page[str] of names and node IDs. Use get_node (with an optional decoder) to load a payload once the ID is known.

Files

Three levels of upload help exist, from most to least hand-holding: upload_file (buffers the whole file in memory, computes the checksum for you), upload_file_chunked (streams arbitrarily-sized chunks without buffering the whole file — you supply the checksum, but it still re-chunks and validates everything else for you), and upload_file_raw (a raw pass-through — you build every protobuf-backed message yourself, with zero validation). The wire contract all three implement is worth understanding even if you only ever call upload_file.

The upload wire contract

Upload is a client-streaming RPC:

• The first message carries the file's metadata — tenant_id, bucket, file_id, size_bytes (the exact total byte count), content_type, checksum, and request_id. Every later message is only read for its chunk field; metadata fields on those later messages are ignored.
• Chunk size is the client's choice, capped at 1 MiB — not a fixed requirement. The server stores each chunk verbatim at its position in the stream and, on download, reads chunks back until it has collected size_bytes bytes in total — it does not assume any particular chunk size when replaying them. A single message's chunk larger than 1 MiB is rejected outright with INVALID_ARGUMENT; anything at or under that cap is fine. upload_file and upload_file_chunked both always emit exactly-1-MiB messages (the last one possibly shorter): 1 MiB is the largest message the server allows, so it is also the fewest possible messages for a given file, and it remains the only chunk size safe against a server older than 1.0.0-rc.16, which reassembled a download from a guessed chunk count instead of the recorded size_bytes. This is also why neither method exposes a chunk-size option.
• checksum must be exactly 32 raw bytes — a SHA-256 digest. Any other length, including empty, is rejected with INVALID_ARGUMENT before a single chunk is read. The server does not verify that the checksum actually matches the uploaded bytes, only that its length is correct.
• The sum of every chunk's bytes across the stream must equal size_bytes exactly, or the server rejects the upload with INVALID_ARGUMENT at the end of the stream — this is what makes size_bytes a value the SDK, and the server on download, can trust, rather than just a caller-supplied claim.
• Re-uploading an existing file_id replaces it, with no error for the duplicate — no separate delete-then-upload dance is required. download_file/stat_file afterward always serve the newest upload.
• Files are capped at the server's limits.max_file_bytes (5 GiB by default, MAX_FILE_BYTES in this SDK). upload_file and upload_file_chunked check this client-side, before any RPC, and raise RociaDbValidationError if it is exceeded.
• An empty file is valid and common: it needs exactly one message (metadata only, empty chunk) and no data messages. upload_file and upload_file_chunked both send it automatically.
• The file only becomes visible (in list_files, stat_file, download_file) once the whole stream has been received and validated. An interrupted stream leaves orphaned chunks that a background GC eventually reclaims; the partial file never appears anywhere, so retrying (with a fresh request_id) is always safe.
• delete_file removes a whole file, regardless of how many chunks it was stored in.

Buffered files

Use the buffered helpers for reasonably sized objects. upload_file computes a SHA-256 checksum of data automatically when checksum is omitted:

payload = b"hello RociaDB"

await client.upload_file(
    "tenant-1", "assets", "manual.txt", payload, content_type="text/plain; charset=utf-8"
)

metadata = await client.stat_file("tenant-1", "assets", "manual.txt")
print(metadata.size_bytes, metadata.content_type)
print(metadata.checksum.hex())  # sha256 hex

data = await client.download_file("tenant-1", "assets", "manual.txt")
await client.delete_file("tenant-1", "assets", "manual.txt")

metadata.checksum is the raw 32-byte digest; hex-encode it for display or comparison, as shown above. If you already have a checksum computed elsewhere, pass it as checksum=... instead — it must be exactly 32 bytes, checked before any network call.

Uploading from a stream without buffering the whole file

upload_file_chunked avoids holding the complete object in memory. The caller must know the total upload size and a precomputed SHA-256 checksum before starting the RPC — unlike upload_file, it cannot hash the data for you, because the checksum has to travel on the first message, before the SDK has read anything from your source:

import hashlib
from pathlib import Path

source = Path("large.bin")
size_bytes = source.stat().st_size

hasher = hashlib.sha256()
with source.open("rb") as f:
    for block in iter(lambda: f.read(65536), b""):
        hasher.update(block)
checksum = hasher.digest()


def read_chunks():
    with source.open("rb") as f:
        while True:
            block = f.read(65536)
            if not block:
                return
            yield block


await client.upload_file_chunked(
    "tenant-1", "assets", "large.bin", size_bytes, checksum, read_chunks()
)

with open("downloaded.bin", "wb") as out:
    async for chunk in client.download_file_stream("tenant-1", "assets", "large.bin"):
        out.write(chunk)

chunks may be a sync Iterable[bytes] or an AsyncIterable[bytes], and its pieces may be any size — read_chunks() above yields 64 KiB blocks — because upload_file_chunked re-buffers internally and only ever writes 1 MiB per outgoing message, per The upload wire contract above. download_file_stream is an async generator: iterate it directly with async for, no await on the call itself.

Porting upload code from the Rust or TypeScript SDK by method name alone is unsafe here: neither upload_file_chunked's nor upload_file_raw's counterpart in those SDKs is the method whose name looks closest — see Parity with the Other SDKs for the naming table before translating upload calls between SDKs.

Raw streaming upload escape hatch

upload_file_raw passes every message straight through to the gRPC call exactly as given — no re-chunking, no checksum validation, and no distinction between the first message and the rest. You are fully responsible for the wire contract described in The upload wire contract above.

import hashlib
from uuid import uuid4

from rociadb_sdk import RawUploadMessage


async def raw_upload():
    payload = b"hello RociaDB"
    yield RawUploadMessage(
        tenant_id="tenant-1",
        bucket="assets",
        file_id="manual.txt",
        size_bytes=len(payload),
        content_type="text/plain",
        checksum=hashlib.sha256(payload).digest(),
        chunk=payload,
        request_id=f"upload_file:{uuid4()}",
    )
    # A larger file would yield further RawUploadMessage instances here,
    # chunk <= 1 MiB each. Every field is required by the dataclass, but the
    # server only reads the non-chunk fields off the first message, so later
    # messages can repeat placeholder values for them.


await client.upload_file_raw(raw_upload())

Getting the chunk size wrong here fails fast with INVALID_ARGUMENT rather than silently corrupting a later download — but a wrong size_bytes total, or a checksum that does not actually match the bytes (the server only checks its length, never its content), can still slip through as an upload that looks successful while carrying bad data. Prefer upload_file or upload_file_chunked above unless you specifically need to hand-build the message stream.

Discovering buckets and files

buckets = await client.list_buckets("tenant-1")
files = await client.list_files("tenant-1", "assets", limit=100)

Tenants

list_tenants is the only RPC that takes no tenant_id: it enumerates every tenant known to the deployment (the registry is filled in implicitly — a tenant appears the first time any RPC mentions it). It lives on its own service so a dedicated authorization policy could be attached to it independently of the data-plane services; today, any authenticated client can call it, including a read-only one. The one credential that cannot is an admin-scoped token — like every other RPC, it gets PERMISSION_DENIED (see Authentication).

from typing import Optional

cursor: Optional[str] = None
while True:
    page = await client.list_tenants(limit=100, cursor=cursor)
    for tenant_id in page.items:
        print(tenant_id)
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

Idempotence and Retries

Every mutating call creates a unique request ID by default. When retrying an operation after a timeout, pass the same request_id so the server can recognize the attempt as the same logical mutation instead of applying it twice:

from uuid import uuid4

request_id = str(uuid4())
await client.put_document("tenant-1", "products", "sku-123", product, request_id=request_id)

A request_id is scoped to (tenant_id, operation, request_id) — replaying the same value against a different operation is not treated as the same mutation. Reusing a put_document call's request_id on a later delete_document call, for instance, does not cancel or replace the earlier write; it is simply a different idempotency key. Markers expire after the server's gc.request_ttl_secs (24 hours by default); a retry older than that window executes again rather than being deduplicated.

Every mutating method except the raw upload escape hatch takes request_id as an optional keyword argument, defaulting to an auto-generated value if omitted; put_nodes and add_edges read it off each NodeInput/EdgeInput item instead, one per batch entry. upload_file_raw has no such default — each RawUploadMessage carries its own request_id explicitly.

Error Handling

All SDK failures raise a subclass of RociaDbError. Catching the base class covers everything; catching a specific subclass narrows to one cause:

Exception	Meaning
RociaDbStatusError	A gRPC call returned a non-OK status. Carries operation, code (a grpc.StatusCode), reason (the server's reason trailing metadata, finer-grained than code alone), and grpc_error (the original grpc.aio.AioRpcError).
RociaDbConnectionError	Failed to connect to, or configure, the endpoint — invalid host, TLS setup, connection refused, missing auth configuration, a non-positive connect timeout.
RociaDbAuthError	Failed to obtain or refresh the OAuth2 bearer token.
RociaDbEncodeError	Failed to JSON-encode a document, node, or edge payload before sending it. Carries context.
RociaDbDecodeError	Failed to JSON-decode a payload received from the server. Carries context.
RociaDbValidationError	A client-side rule was rejected before any network call — a non-positive page limit, a checksum of the wrong length, a file size out of bounds, a partial node_label/node_graph pair, and so on.

Only RociaDbStatusError ever carries a gRPC code/reason; the other subclasses are always raised client-side, before any RPC. Narrow on the exception type first to handle a whole category (every validation error alike, say), and on code/reason for gRPC-specific branching:

import grpc

from rociadb_sdk import RociaDbStatusError

try:
    await client.get_document("tenant-1", "products", "missing")
except RociaDbStatusError as error:
    if error.code == grpc.StatusCode.NOT_FOUND:
        print("document does not exist")
    else:
        raise

Catch the RociaDbError base instead of a specific subclass for a catch-all across every possible SDK failure.

gRPC code	reason	When
INVALID_ARGUMENT	invalid_argument	Missing/malformed field, limit out of bounds, unreadable cursor, invalid JSON
NOT_FOUND	not_found	Document, node, edge, or file does not exist
ALREADY_EXISTS	already_exists	A uniqueness conflict
PERMISSION_DENIED	permission_denied	Insufficient scope — see Authentication
UNAUTHENTICATED	unauthenticated	Token missing, expired, malformed, or from another issuer — see Authentication
INTERNAL	internal	Storage-layer failure

error.is_unauthenticated() and error.is_permission_denied() are shorthands for checking code against those two values; they are defined on the RociaDbError base (returning False) so either is safe to call on any caught error, as shown in Authentication.

API Coverage

Service	RPC	SDK method
Document	PutDoc	put_document, create_document
Document	GetDoc	get_document
Document	DeleteDoc	delete_document
Document	FindByField	find_documents_by_field
Document	ListDoc	list_documents
Document	QueryDoc	query_documents
Document	ListCollections	list_collections
Graph	PutNode	put_node, put_nodes
Graph	GetNode	get_node
Graph	AddEdge	add_edge, add_edges
Graph	DeleteEdge	delete_edge
Graph	NeighborsOut	neighbors_out, get_outgoing_neighbor_nodes
Graph	NeighborsIn	neighbors_in, get_incoming_neighbor_nodes
Graph	ListGraphs	list_graphs
Graph	ListNodes	list_nodes
File	Upload	upload_file, upload_file_chunked, upload_file_raw
File	Download	download_file, download_file_stream
File	Stat	stat_file
File	Delete	delete_file
File	ListBuckets	list_buckets
File	ListFiles	list_files
Tenant	ListTenants	list_tenants

Parity with the Other SDKs

This SDK, the Rust SDK (rociadb-core-sdk-rust), and the TypeScript SDK (rociadb-core-sdk-ts) cover the same 22 RPCs against the same server and are maintained to the same standard: every capability available in one is available in all three. None imitates another's syntax — this package stays snake_case/exception-idiomatic Python, the Rust crate stays snake_case/Result-idiomatic Rust, the TypeScript package stays camelCase/Promise-idiomatic TypeScript — but a piece of client code should always have a mechanical translation from any one SDK to either of the others. Parity is about what you can do, not about identical method names. Because this SDK and the Rust SDK both use snake_case, most method names already match character for character between them (put_nodes, get_outgoing_neighbor_nodes, list_documents, and so on); translating to or from the TypeScript SDK is usually just the camelCase transform (putNodes, getOutgoingNeighborNodes, listDocuments). The handful of places where a name does not translate mechanically are the naming table below.

Capability	Python (this SDK)	Rust	TypeScript	Note
Assisted streaming upload — re-chunks to the 1 MiB wire contract, validates the total, caller supplies the checksum	upload_file_chunked	upload_file_chunked	uploadFileStream	Python reuses Rust's unambiguous name. Names do not correspond to TypeScript's uploadFileStream — see the naming trap below.
Raw streaming upload — zero validation, caller builds every message	upload_file_raw	upload_file_stream	uploadFileRaw	Python reuses TypeScript's unambiguous name. This is Rust's upload_file_stream, the mirror image of the row above.
Idempotency key scoped to a create_document/createDocument call's document write only (the graph node binding keeps its own auto-generated key)	create_document(..., request_id=...) — a keyword argument	create_document_with_request_id — a sibling method	createDocument(..., { requestId }) — an options-object field	Same capability, three idiomatic shapes for an optional named argument in each language.
Releasing the connection and any cached auth state	close() / async with (owns its channel outright; must not be reused after)	Drop the last live RociaDbClient clone (Clone + one shared channel; no method)	close() (owns its channel outright, same model as Python)	Rust's ownership model shares one channel across every clone, so there is no explicit release method to call — dropping the last clone is the release point.
Marking the cached token stale without blocking on a refresh	invalidate_auth_token() — wakes a background refresh task; the cached token is still used until that refresh completes	invalidate_auth_token() / TokenManager::request_refresh — same background-task design	invalidateToken() — synchronously drops the cached token; the next call fetches inline	Python mirrors Rust's background-refresh design (see Authentication) rather than TypeScript's on-demand-near-expiry design, so the method name matches Rust while the exact timing differs from TypeScript's synchronous drop.
Discriminating why an error happened	RociaDbError subclass hierarchy — isinstance/except on RociaDbStatusError, RociaDbConnectionError, etc.	RociaDbError — a match-able enum: Status { .. } / Connection { .. } / ...	RociaDbError — one class with a kind: "status" | "connection" | ... field	Three idiomatic shapes for the same closed set of six causes: a subclass hierarchy, a sum type, and a discriminated union.
Escape hatch to the raw generated protobuf/gRPC types, to build a custom client against the same .proto	(none — rociadb_sdk._pb is private; every public method returns this package's own dataclasses)	the pb module (#[doc(hidden)] pub mod pb; a few generated types are re-exported individually at the crate root)	the rocia-db-sdk/proto subpath export	Python deliberately keeps its generated stubs fully private — there is no equivalent escape hatch in this SDK by design.

The upload naming trap, spelled out: upload_file_chunked (this SDK, same name as Rust) and uploadFileStream (TypeScript) are the same capability — the middle tier that re-chunks and validates for you (see Uploading from a stream without buffering the whole file). upload_file_raw (this SDK, same name as TypeScript) and upload_file_stream (Rust) are also the same capability — the raw, zero-validation escape hatch (see Raw streaming upload escape hatch). upload_file_chunked and upload_file_stream are not each other's counterpart despite the near-identical name: reading either name in isolation and guessing at the other SDK's equivalent by ear silently swaps which tier you land on. Always cross-check against the table above.

Two capabilities are intentionally kept on one or two sides without a mirror everywhere: ApiKeyInterceptor (Rust only — it validates an incoming API key, so it serves building a server or a test double, not talking to RociaDB, which puts it out of scope for any client SDK), and having both a fluent RociaDbBuilder and a direct one-call constructor (RociaDbClient.connect) — Python and TypeScript both offer this pair; Rust offers only the builder, since a second entry point there would add an API to maintain for no new capability.

Development and Testing

The project is managed with uv. From a clone of this repository:

uv sync
uv run --python 3.10 pytest -q
uv run --python 3.14 pytest -q      # repeat for 3.11, 3.12, 3.13 as needed
uv run --python 3.10 ruff check .
uv run --python 3.10 ruff format --check .
uv run --python 3.10 mypy

The protobuf/gRPC stubs under src/rociadb_sdk/_pb/ are generated, vendored code — never edit them by hand. After changing proto/upstream/v1/upstream.proto (itself a byte-identical copy of the server's .proto file, never edited independently), regenerate them with:

uv run --python 3.10 python scripts/generate_proto.py

Licensed under Apache-2.0.