Skip to content

Result Backends

ResultMiddleware

ResultMiddleware

Bases: BaseMiddleware

Stores handler return values in a result backend.

Keyed by the message's correlation_id. If no correlation_id, skips storage.

Source code in src/rabbitkit/results/middleware.py 
 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
class ResultMiddleware(BaseMiddleware):
    """Stores handler return values in a result backend.

    Keyed by the message's correlation_id. If no correlation_id, skips storage.
    """

    def __init__(self, backend: ResultBackend[Any], serializer: Any | None = None, ttl: int = 3600) -> None:
        self._backend = backend
        self._serializer = serializer
        self._ttl = ttl

    def _serialize(self, result: Any) -> bytes:
        """Encode a handler's return value for storage.

        H13: no ``default=str`` fallback — a non-JSON-native object (custom
        class, ``exception``, etc.) raises ``TypeError`` here instead of
        being silently stringified into a lossy blob indistinguishable from
        a real result. An exception specifically gets an explicit,
        marked error envelope (``__rabbitkit_error__``) rather than either
        of those, since "the handler returned an exception object as data"
        is a legitimate pattern worth preserving in a decodable, unambiguous
        shape. Pass an explicit ``serializer=`` to support other
        non-JSON-native result types.
        """
        if isinstance(result, bytes):
            return result
        if self._serializer is not None and hasattr(self._serializer, "encode"):
            return self._serializer.encode(result)  # type: ignore[no-any-return]
        if isinstance(result, BaseException):
            envelope = {
                "__rabbitkit_error__": True,
                "type": type(result).__qualname__,
                "message": str(result),
            }
            return json.dumps(envelope).encode("utf-8")
        return json.dumps(result).encode("utf-8")

    def consume_scope(self, call_next: Any, message: RabbitMessage) -> Any:
        result = call_next(message)
        if result is not None and message.correlation_id:
            self._backend.store(message.correlation_id, self._serialize(result), self._ttl)
        return result

    async def consume_scope_async(self, call_next: Any, message: RabbitMessage) -> Any:
        result = await call_next(message)
        if result is not None and message.correlation_id:
            await self._backend.store_async(message.correlation_id, self._serialize(result), self._ttl)
        return result

ResultBackend Protocol

ResultBackend

Bases: Protocol[T]

Protocol for result storage backends.

Generic in T (the stored result type) so the type flows from store into fetch. RedisResultBackend stores bytes, so it satisfies ResultBackend[bytes].

Source code in src/rabbitkit/results/backend.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
@runtime_checkable
class ResultBackend(Protocol[T]):
    """Protocol for result storage backends.

    Generic in ``T`` (the stored result type) so the type flows from
    ``store`` into ``fetch``.  ``RedisResultBackend`` stores ``bytes``, so it
    satisfies ``ResultBackend[bytes]``.
    """

    def store(self, correlation_id: str, result: T, ttl: int = 3600) -> None: ...
    def fetch(self, correlation_id: str, timeout: float = 5.0) -> T | None: ...
    async def store_async(self, correlation_id: str, result: T, ttl: int = 3600) -> None: ...
    async def fetch_async(self, correlation_id: str, timeout: float = 5.0) -> T | None: ...

RedisResultBackend

RedisResultBackend

Redis-based result backend using GET/SET with TTL.

Stores results as raw bytes, so it satisfies ResultBackend[bytes].

Source code in src/rabbitkit/results/backend.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class RedisResultBackend:
    """Redis-based result backend using GET/SET with TTL.

    Stores results as raw bytes, so it satisfies ``ResultBackend[bytes]``.
    """

    def __init__(self, redis_client: Any, key_prefix: str = "rabbitkit:result:") -> None:
        self._redis = redis_client
        self._prefix = key_prefix

    def _key(self, correlation_id: str) -> str:
        return f"{self._prefix}{correlation_id}"

    def store(self, correlation_id: str, result: bytes, ttl: int = 3600) -> None:
        self._redis.set(self._key(correlation_id), result, ex=ttl)

    def fetch(self, correlation_id: str, timeout: float = 5.0) -> bytes | None:
        return self._redis.get(self._key(correlation_id))  # type: ignore[no-any-return]

    async def store_async(self, correlation_id: str, result: bytes, ttl: int = 3600) -> None:
        await self._redis.set(self._key(correlation_id), result, ex=ttl)

    async def fetch_async(self, correlation_id: str, timeout: float = 5.0) -> bytes | None:
        return await self._redis.get(self._key(correlation_id))  # type: ignore[no-any-return]