udb 0.3.0

Universal Data Broker — a Rust gRPC broker over multiple databases (Postgres, MySQL, SQLite, MongoDB, ClickHouse, Cassandra, MSSQL, Redis, Qdrant, S3, Neo4j, …) with per-tenant RLS, 2PC, sagas, and CDC.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

from __future__ import annotations

from dataclasses import dataclass, field, replace
import os
from typing import Iterable, Sequence

from udb.entity.v1 import types_pb2

UDB_PROTOCOL_VERSION = "1.0.0"


@dataclass(frozen=True)
class Metadata:
    """Request metadata required by the UDB broker."""

    tenant_id: str
    purpose: str
    correlation_id: str
    scopes: Sequence[str] = field(default_factory=tuple)
    service_identity: str = "python.app"
    user_id: str = ""
    project_id: str = "default"
    client_catalog_version: str = UDB_PROTOCOL_VERSION
    consistency: str = ""
    target_backend: str = ""
    target_instance: str = ""
    routing_policy: str = ""
    primary_read: bool = False
    max_replica_lag_ms: int = 0
    eventual_consistency_allowed: bool = False
    read_fence_json: str = ""
    trace_id: str = ""

    @classmethod
    def from_env(
        cls,
        *,
        tenant_id: str | None = None,
        purpose: str | None = None,
        correlation_id: str | None = None,
        scopes: Sequence[str] | None = None,
        prefix: str = "UDB_",
    ) -> "Metadata":
        """Build metadata from environment variables with explicit overrides."""

        def env(name: str, default: str = "") -> str:
            return os.getenv(f"{prefix}{name}", default)

        raw_scopes = env("SCOPES")
        parsed_scopes = tuple(
            scope.strip() for scope in raw_scopes.split(",") if scope.strip()
        )
        return cls(
            tenant_id=tenant_id or env("TENANT_ID"),
            user_id=env("USER_ID"),
            purpose=purpose or env("PURPOSE", "python.request"),
            correlation_id=correlation_id or env("CORRELATION_ID"),
            scopes=tuple(scopes) if scopes is not None else parsed_scopes,
            service_identity=env("SERVICE_IDENTITY", "python.app"),
            project_id=env("PROJECT_ID", "default"),
            client_catalog_version=env(
                "CLIENT_CATALOG_VERSION", UDB_PROTOCOL_VERSION
            ),
            consistency=env("CONSISTENCY"),
            target_backend=env("TARGET_BACKEND"),
            target_instance=env("TARGET_INSTANCE"),
            routing_policy=env("ROUTING_POLICY"),
            primary_read=env("PRIMARY_READ").lower() == "true",
            max_replica_lag_ms=int(env("MAX_REPLICA_LAG_MS", "0") or "0"),
            eventual_consistency_allowed=(
                env("EVENTUAL_CONSISTENCY_ALLOWED").lower() == "true"
            ),
            read_fence_json=env("READ_FENCE_JSON"),
            trace_id=env("TRACE_ID"),
        )

    def to_grpc_metadata(self) -> tuple[tuple[str, str], ...]:
        headers: list[tuple[str, str]] = [
            ("x-tenant-id", self.tenant_id),
            ("x-user-id", self.user_id),
            ("x-purpose", self.purpose),
            ("x-correlation-id", self.correlation_id),
            ("x-scopes", ",".join(self.scopes)),
            ("x-service-identity", self.service_identity),
            ("x-udb-project-id", self.project_id),
            ("x-udb-client-catalog-version", self.client_catalog_version),
        ]
        optional = {
            "x-udb-consistency": self.consistency,
            "x-udb-target-backend": self.target_backend,
            "x-udb-target-instance": self.target_instance,
            "x-udb-routing-policy": self.routing_policy,
            "x-udb-read-fence": self.read_fence_json,
            "x-udb-max-replica-lag-ms": (
                str(self.max_replica_lag_ms) if self.max_replica_lag_ms else ""
            ),
            "x-udb-primary-read": "true" if self.primary_read else "",
            "x-udb-eventual-consistency-allowed": (
                "true" if self.eventual_consistency_allowed else ""
            ),
        }
        headers.extend((key, value) for key, value in optional.items() if value)
        return tuple(headers)

    def grpc_metadata(self) -> Iterable[tuple[str, str]]:
        """Backward-compatible alias for older examples."""

        return self.to_grpc_metadata()

    def to_request_context(self) -> types_pb2.RequestContext:
        return types_pb2.RequestContext(
            tenant_id=self.tenant_id,
            user_id=self.user_id,
            purpose=self.purpose,
            correlation_id=self.correlation_id,
            scopes=list(self.scopes),
            service_identity=self.service_identity,
            trace_id=self.trace_id,
            target_backend=self.target_backend,
            target_instance=self.target_instance,
            routing_policy=self.routing_policy,
            primary_read=self.primary_read,
            max_replica_lag_ms=self.max_replica_lag_ms,
            eventual_consistency_allowed=self.eventual_consistency_allowed,
            read_fence_json=self.read_fence_json,
        )

    def with_purpose(self, purpose: str) -> "Metadata":
        return replace(self, purpose=purpose)

    def with_scopes(self, scopes: Sequence[str]) -> "Metadata":
        return replace(self, scopes=tuple(scopes))

    def with_project_id(self, project_id: str) -> "Metadata":
        return replace(self, project_id=project_id)

    def with_read_fence(self, read_fence_json: str) -> "Metadata":
        return replace(self, read_fence_json=read_fence_json)