udb-client 0.3.1

Python SDK for the UDB DataBroker gRPC API

UDB Python SDK

udb-client is the Python client for UDB. It gives you sync and async clients, metadata handling, common CRUD helpers, optional Pydantic models, raw access to every broker RPC, and a version-matched udb CLI launcher.

Install

pip install udb-client==0.3.1

Optional validated command models:

pip install "udb-client[pydantic]==0.3.1"

Runtime: Python 3.10+

Export UDB Protos For Your App

After installation, the udb command is available. Use it from your app project to copy UDB's annotation and broker protos into your own proto tree:

udb proto export

Then your project schemas can import:

import "udb/core/common/v1/db.proto";

Run udb proto export again whenever you upgrade the SDK/CLI. It keeps the UDB annotation protos and your installed CLI version together.

Basic CRUD

from udb_client import Metadata, UdbClient, decode_records

meta = Metadata(
    tenant_id="acme",
    user_id="user-1",
    purpose="billing.api",
    correlation_id="request-001",
    scopes=("udb:read", "udb:write"),
    service_identity="billing-service",
    project_id="billing",
)

with UdbClient("127.0.0.1:50051", meta) as udb:
    udb.upsert(
        message_type="acme.billing.v1.Customer",
        record={
            "customer_id": "cus_001",
            "tenant_id": "acme",
            "name": "Ada Lovelace",
            "email": "[e-mail]",
        },
        conflict_fields=("customer_id",),
        return_record=True,
    )

    rows = udb.select(
        message_type="acme.billing.v1.Customer",
        filter={"customer_id": "cus_001"},
        limit=1,
    )
    print(decode_records(rows))

Async

from udb_client import Metadata, UdbAsyncClient

meta = Metadata(
    tenant_id="acme",
    purpose="billing.worker",
    correlation_id="job-42",
    scopes=("udb:read", "udb:write"),
    service_identity="billing-worker",
    project_id="billing",
)

async with UdbAsyncClient("127.0.0.1:50051", meta) as udb:
    await udb.upsert(
        message_type="acme.billing.v1.Customer",
        record={"customer_id": "cus_002", "tenant_id": "acme"},
        conflict_fields=("customer_id",),
    )

Authz Check

from udb_client.auth import UdbAuthClient
from udb.core.authz.services.v1 import core_pb2 as authz

with UdbAuthClient("127.0.0.1:50051", meta) as auth:
    allowed, decision = auth.can(
        authz.ResourceRef(message_type="acme.billing.v1.Customer"),
        "read",
    )

Full API Access

For broker APIs without a convenience method, use client.call("RpcName", request) for unary RPCs with metadata/error handling, or client.stub for raw streaming and advanced calls.

The generated protobuf modules are included for request/response types:

from udb.entity.v1 import types_pb2
from udb.services.v1 import data_broker_pb2_grpc

Local SDK Development

Consumers do not need this. Use it only when editing the SDK itself:

cd sdk/python
python -m pip install -e ".[dev]"
python scripts/generate_protos.py
pytest
pyrefly check