Add typed gRPC exceptions and capture trailing metadata

Add TimeoutError, ResourceExhaustedError, CancelledError, and UnimplementedError
to the gRPC status code map so callers can catch by type instead of inspecting .code.

DecreeError now captures trailing_metadata from the RpcError and exposes a
retry_after timedelta parsed from google.rpc.RetryInfo packed in the
grpc-status-details-bin trailer.

Closes #54

Author: zeevdr

sdk/pyproject.toml

@@ -75,6 +75,10 @@ ignore_errors = true
 module = "grpc.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "google.rpc.*"
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"

sdk/src/opendecree/__init__.py

@@ -14,15 +14,19 @@
 from opendecree.client import ConfigClient
 from opendecree.errors import (
     AlreadyExistsError,
+    CancelledError,
     ChecksumMismatchError,
     DecreeError,
     IncompatibleServerError,
     InvalidArgumentError,
     LockedError,
     NotFoundError,
     PermissionDeniedError,
+    ResourceExhaustedError,
+    TimeoutError,
     TypeMismatchError,
     UnavailableError,
+    UnimplementedError,
 )
 from opendecree.types import Change, ConfigValue, FieldUpdate, ServerVersion
 from opendecree.watcher import ConfigWatcher, WatchedField
@@ -35,6 +39,7 @@
     "AsyncConfigClient",
     "AsyncConfigWatcher",
     "AsyncWatchedField",
+    "CancelledError",
     "Change",
     "ChecksumMismatchError",
     "ConfigClient",
@@ -47,10 +52,13 @@
     "LockedError",
     "NotFoundError",
     "PermissionDeniedError",
+    "ResourceExhaustedError",
     "RetryConfig",
     "ServerVersion",
+    "TimeoutError",
     "TypeMismatchError",
     "UnavailableError",
+    "UnimplementedError",
     "WatchedField",
     "__version__",
 ]

sdk/src/opendecree/errors.py

@@ -5,21 +5,28 @@
 
 from __future__ import annotations
 
+import datetime
+
 import grpc
+import grpc.aio
+from google.rpc import error_details_pb2, status_pb2
 
 
 class DecreeError(Exception):
     """Base exception for all OpenDecree SDK errors."""
 
-    def __init__(self, message: str, code: grpc.StatusCode | None = None) -> None:
-        """Create a DecreeError.
-
-        Args:
-            message: Human-readable error description.
-            code: The gRPC status code that caused this error, if any.
-        """
+    def __init__(
+        self,
+        message: str,
+        code: grpc.StatusCode | None = None,
+        *,
+        trailing_metadata: grpc.aio.Metadata | None = None,
+        retry_after: datetime.timedelta | None = None,
+    ) -> None:
         super().__init__(message)
         self.code = code
+        self.trailing_metadata = trailing_metadata
+        self.retry_after = retry_after
 
 
 class NotFoundError(DecreeError):
@@ -58,6 +65,22 @@ class TypeMismatchError(DecreeError):
     """Raised when a typed getter receives a value of the wrong type."""
 
 
+class TimeoutError(DecreeError):
+    """Raised when the operation deadline was exceeded."""
+
+
+class ResourceExhaustedError(DecreeError):
+    """Raised when a resource quota or rate limit is exceeded."""
+
+
+class CancelledError(DecreeError):
+    """Raised when the operation was cancelled."""
+
+
+class UnimplementedError(DecreeError):
+    """Raised when the server has not implemented the operation."""
+
+
 _STATUS_MAP: dict[grpc.StatusCode, type[DecreeError]] = {
     grpc.StatusCode.NOT_FOUND: NotFoundError,
     grpc.StatusCode.ALREADY_EXISTS: AlreadyExistsError,
@@ -67,9 +90,27 @@ class TypeMismatchError(DecreeError):
     grpc.StatusCode.PERMISSION_DENIED: PermissionDeniedError,
     grpc.StatusCode.UNAUTHENTICATED: PermissionDeniedError,
     grpc.StatusCode.UNAVAILABLE: UnavailableError,
+    grpc.StatusCode.DEADLINE_EXCEEDED: TimeoutError,
+    grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhaustedError,
+    grpc.StatusCode.CANCELLED: CancelledError,
+    grpc.StatusCode.UNIMPLEMENTED: UnimplementedError,
 }
 
 
+def _parse_retry_after(metadata: grpc.aio.Metadata) -> datetime.timedelta | None:
+    for key, value in metadata:
+        if key != "grpc-status-details-bin" or not isinstance(value, bytes):
+            continue
+        rpc_status = status_pb2.Status()
+        rpc_status.ParseFromString(value)
+        for detail in rpc_status.details:
+            retry_info = error_details_pb2.RetryInfo()
+            if detail.Unpack(retry_info):
+                d = retry_info.retry_delay
+                return datetime.timedelta(seconds=d.seconds, microseconds=d.nanos // 1000)
+    return None
+
+
 def map_grpc_error(err: grpc.RpcError) -> DecreeError:
     """Convert a gRPC ``RpcError`` to a typed ``DecreeError``.
 
@@ -78,5 +119,7 @@ def map_grpc_error(err: grpc.RpcError) -> DecreeError:
     """
     code = err.code()
     details = err.details()
+    trailing: grpc.aio.Metadata | None = getattr(err, "trailing_metadata", lambda: None)()
+    retry_after = _parse_retry_after(trailing) if trailing else None
     exc_class = _STATUS_MAP.get(code, DecreeError)
-    return exc_class(details or str(err), code)
+    return exc_class(details or str(err), code, trailing_metadata=trailing, retry_after=retry_after)

sdk/tests/conftest.py

@@ -13,7 +13,12 @@ class FakeRpcError(grpc.aio.AioRpcError):
     sync and async error handling.
     """
 
-    def __init__(self, code: grpc.StatusCode, details: str = "test") -> None:
+    def __init__(
+        self,
+        code: grpc.StatusCode,
+        details: str = "test",
+        trailing_metadata: grpc.aio.Metadata | None = None,
+    ) -> None:
         # AioRpcError.__init__ expects (code, initial_metadata, trailing_metadata,
         # details, debug_error_string).
-        super().__init__(code, None, None, details, None)  # type: ignore[arg-type]
+        super().__init__(code, None, trailing_metadata, details, None)  # type: ignore[arg-type]

sdk/tests/test_errors.py

@@ -1,20 +1,39 @@
 """Tests for error mapping."""
 
+import datetime
+
 import grpc
+import grpc.aio
+from google.protobuf import any_pb2, duration_pb2
+from google.rpc import error_details_pb2, status_pb2
 
 from opendecree.errors import (
     AlreadyExistsError,
+    CancelledError,
     ChecksumMismatchError,
     DecreeError,
     LockedError,
     NotFoundError,
     PermissionDeniedError,
+    ResourceExhaustedError,
+    TimeoutError,
     UnavailableError,
+    UnimplementedError,
     map_grpc_error,
 )
 from tests.conftest import FakeRpcError
 
 
+def _make_retry_metadata(seconds: int, nanos: int = 0) -> grpc.aio.Metadata:
+    retry_info = error_details_pb2.RetryInfo(
+        retry_delay=duration_pb2.Duration(seconds=seconds, nanos=nanos)
+    )
+    detail = any_pb2.Any()
+    detail.Pack(retry_info)
+    rpc_status = status_pb2.Status(details=[detail])
+    return grpc.aio.Metadata(("grpc-status-details-bin", rpc_status.SerializeToString()))
+
+
 def test_not_found():
     err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND, "gone"))
     assert isinstance(err, NotFoundError)
@@ -61,3 +80,55 @@ def test_unknown_code_falls_back():
 def test_empty_details():
     err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND, ""))
     assert isinstance(err, NotFoundError)
+
+
+def test_deadline_exceeded_maps_to_timeout():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.DEADLINE_EXCEEDED))
+    assert isinstance(err, TimeoutError)
+    assert err.code == grpc.StatusCode.DEADLINE_EXCEEDED
+
+
+def test_resource_exhausted():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.RESOURCE_EXHAUSTED))
+    assert isinstance(err, ResourceExhaustedError)
+
+
+def test_cancelled():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.CANCELLED))
+    assert isinstance(err, CancelledError)
+
+
+def test_unimplemented():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNIMPLEMENTED))
+    assert isinstance(err, UnimplementedError)
+
+
+def test_trailing_metadata_captured():
+    meta = grpc.aio.Metadata(("x-custom", "value"))
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNAVAILABLE, trailing_metadata=meta))
+    assert err.trailing_metadata is not None
+    assert dict(err.trailing_metadata).get("x-custom") == "value"
+
+
+def test_no_trailing_metadata():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND))
+    assert err.trailing_metadata is None
+    assert err.retry_after is None
+
+
+def test_retry_info_parsed():
+    meta = _make_retry_metadata(seconds=5, nanos=500_000_000)
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.RESOURCE_EXHAUSTED, trailing_metadata=meta))
+    assert err.retry_after == datetime.timedelta(seconds=5, microseconds=500_000)
+
+
+def test_retry_info_nanos_precision():
+    meta = _make_retry_metadata(seconds=0, nanos=250_000_000)
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.DEADLINE_EXCEEDED, trailing_metadata=meta))
+    assert err.retry_after == datetime.timedelta(microseconds=250_000)
+
+
+def test_no_retry_info_in_metadata():
+    meta = grpc.aio.Metadata(("x-other", "val"))
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNAVAILABLE, trailing_metadata=meta))
+    assert err.retry_after is None