Skip to content

domain.protocols.cache_protocol

src.domain.protocols.cache_protocol

Cache protocol for domain layer.

This module defines the cache interface that the domain needs, without knowing about any specific implementation. Infrastructure adapters implement this protocol to provide caching functionality.

Architecture: - Protocol-based - uses structural typing - TypedDict for internal cache metadata - All operations return Result types - No framework dependencies in domain layer

Classes

CacheEntry

Bases: TypedDict

Internal cache entry metadata.

Used internally by cache adapters to track cache state. NOT exposed to domain services (they just get string values).

Attributes:

Name Type Description
key ReadOnly[str]

Cache key (immutable - ReadOnly).
value str

Cached value (string).
ttl int | None

Time to live in seconds (None = no expiration).
created_at ReadOnly[float]

Unix timestamp when entry was created (immutable - ReadOnly).
expires_at float | None

Unix timestamp when entry expires (None if no TTL).
Source code in src/domain/protocols/cache_protocol.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class CacheEntry(TypedDict):
    """Internal cache entry metadata.

    Used internally by cache adapters to track cache state.
    NOT exposed to domain services (they just get string values).

    Attributes:
        key: Cache key (immutable - ReadOnly).
        value: Cached value (string).
        ttl: Time to live in seconds (None = no expiration).
        created_at: Unix timestamp when entry was created (immutable - ReadOnly).
        expires_at: Unix timestamp when entry expires (None if no TTL).
    """

    key: ReadOnly[str]
    value: str
    ttl: int | None
    created_at: ReadOnly[float]
    expires_at: float | None

CacheProtocol

Bases: Protocol

Cache protocol - what domain needs from cache.

Defines caching operations using Protocol (structural typing). Infrastructure adapters implement this without inheritance.

All operations return Result types for error handling. Fail-open strategy: cache failures should not break core functionality.

Source code in src/domain/protocols/cache_protocol.py 
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
class CacheProtocol(Protocol):
    """Cache protocol - what domain needs from cache.

    Defines caching operations using Protocol (structural typing).
    Infrastructure adapters implement this without inheritance.

    All operations return Result types for error handling.
    Fail-open strategy: cache failures should not break core functionality.
    """

    async def get(self, key: str) -> Result[str | None, DomainError]:
        """Get value from cache.

        Args:
            key: Cache key.

        Returns:
            Result with value if found, None if not found, or CacheError.

        Example:
            result = await cache.get("user:123")
            match result:
                case Success(value) if value:
                    # Key found
                    user_data = json.loads(value)
                case Success(None):
                    # Key not found (cache miss)
                    pass
                case Failure(error):
                    # Cache error - fail open
                    logger.warning("Cache get failed", error=error)
        """
        ...

    async def get_json(self, key: str) -> Result[dict[str, Any] | None, DomainError]:
        """Get JSON value from cache.

        Convenience method that deserializes JSON automatically.

        Args:
            key: Cache key.

        Returns:
            Result with parsed dict if found, None if not found, or CacheError.

        Example:
            result = await cache.get_json("user:123")
            match result:
                case Success(data) if data:
                    user_id = data["id"]
                case Success(None):
                    # Cache miss
                    pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def set(
        self,
        key: str,
        value: str,
        ttl: int | None = None,
    ) -> Result[None, DomainError]:
        """Set value in cache.

        Args:
            key: Cache key.
            value: Value to cache (string).
            ttl: Time to live in seconds (None = no expiration).

        Returns:
            Result with None on success, or CacheError.

        Example:
            result = await cache.set(
                "user:123",
                json.dumps(user_data),
                ttl=3600  # 1 hour
            )
            match result:
                case Success(_):
                    # Cached successfully
                    pass
                case Failure(_):
                    # Fail open - continue without cache
                    pass
        """
        ...

    async def set_json(
        self,
        key: str,
        value: dict[str, Any],
        ttl: int | None = None,
    ) -> Result[None, DomainError]:
        """Set JSON value in cache.

        Convenience method that serializes dict to JSON automatically.

        Args:
            key: Cache key.
            value: Dict to cache (will be JSON serialized).
            ttl: Time to live in seconds (None = no expiration).

        Returns:
            Result with None on success, or CacheError.

        Example:
            result = await cache.set_json(
                "user:123",
                {"id": "123", "email": "user@example.com"},
                ttl=1800  # 30 minutes
            )
        """
        ...

    async def delete(self, key: str) -> Result[bool, DomainError]:
        """Delete key from cache.

        Args:
            key: Cache key to delete.

        Returns:
            Result with True if key was deleted, False if key didn't exist,
            or CacheError.

        Example:
            result = await cache.delete("user:123")
            match result:
                case Success(True):
                    # Key was deleted
                    pass
                case Success(False):
                    # Key didn't exist
                    pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def exists(self, key: str) -> Result[bool, DomainError]:
        """Check if key exists in cache.

        Args:
            key: Cache key to check.

        Returns:
            Result with True if key exists, False if not, or CacheError.

        Example:
            result = await cache.exists("user:123")
            match result:
                case Success(True):
                    # Key exists
                    pass
                case Success(False):
                    # Key doesn't exist
                    pass
                case Failure(_):
                    # Fail open - assume doesn't exist
                    pass
        """
        ...

    async def expire(self, key: str, seconds: int) -> Result[bool, DomainError]:
        """Set expiration time on key.

        Args:
            key: Cache key.
            seconds: Seconds until expiration.

        Returns:
            Result with True if timeout was set, False if key doesn't exist,
            or CacheError.

        Example:
            result = await cache.expire("session:abc", 1800)
            match result:
                case Success(True):
                    # Expiration set
                    pass
                case Success(False):
                    # Key doesn't exist
                    pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def ttl(self, key: str) -> Result[int | None, DomainError]:
        """Get time to live for key.

        Args:
            key: Cache key.

        Returns:
            Result with seconds until expiration, None if no TTL or key doesn't
            exist, or CacheError.

        Example:
            result = await cache.ttl("session:abc")
            match result:
                case Success(seconds) if seconds:
                    # Key expires in `seconds`
                    pass
                case Success(None):
                    # No TTL or key doesn't exist
                    pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def increment(self, key: str, amount: int = 1) -> Result[int, DomainError]:
        """Increment numeric value in cache (atomic).

        If key doesn't exist, it's created with value = amount.

        Args:
            key: Cache key.
            amount: Amount to increment by (default: 1).

        Returns:
            Result with new value after increment, or CacheError.

        Example (rate limiting):
            result = await cache.increment(f"rate_limit:{user_id}:{endpoint}")
            match result:
                case Success(count) if count == 1:
                    # First request - set expiration
                    await cache.expire(key, 60)
                case Success(count) if count > 100:
                    # Rate limit exceeded
                    raise RateLimitError()
                case Failure(_):
                    # Fail open - allow request
                    pass
        """
        ...

    async def decrement(self, key: str, amount: int = 1) -> Result[int, DomainError]:
        """Decrement numeric value in cache (atomic).

        Args:
            key: Cache key.
            amount: Amount to decrement by (default: 1).

        Returns:
            Result with new value after decrement, or CacheError.
        """
        ...

    async def flush(self) -> Result[None, DomainError]:
        """Flush all keys from cache.

        WARNING: Use with extreme caution! This clears ALL cache data.
        Should only be used in tests or maintenance windows.

        Returns:
            Result with None on success, or CacheError.

        Example (tests only):
            await cache.flush()  # Clear cache between tests
        """
        ...

    async def ping(self) -> Result[bool, DomainError]:
        """Check cache connectivity (health check).

        Returns:
            Result with True if cache is reachable, or CacheError.

        Example:
            result = await cache.ping()
            match result:
                case Success(True):
                    # Cache is healthy
                    pass
                case Failure(_):
                    # Cache unreachable
                    logger.error("Cache health check failed")
        """
        ...

    async def delete_pattern(self, pattern: str) -> Result[int, DomainError]:
        """Delete all keys matching pattern.

        Args:
            pattern: Glob-style pattern (e.g., "authz:user123:*").

        Returns:
            Result with number of keys deleted, or CacheError.

        Example:
            result = await cache.delete_pattern("session:user123:*")
            match result:
                case Success(count):
                    logger.info(f"Deleted {count} keys")
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def get_many(
        self, keys: list[str]
    ) -> Result[dict[str, str | None], DomainError]:
        """Get multiple values at once (batch operation).

        More efficient than multiple get() calls - uses Redis MGET.

        Args:
            keys: List of cache keys to retrieve.

        Returns:
            Result with dict mapping keys to values (None for missing keys),
            or CacheError.

        Example:
            result = await cache.get_many(["user:1", "user:2", "user:3"])
            match result:
                case Success(data):
                    for key, value in data.items():
                        if value is not None:
                            # Process cached value
                            pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...

    async def set_many(
        self,
        mapping: dict[str, str],
        ttl: int | None = None,
    ) -> Result[None, DomainError]:
        """Set multiple values at once (batch operation).

        More efficient than multiple set() calls - uses Redis pipeline.

        Args:
            mapping: Dict of key->value pairs to cache.
            ttl: Time to live in seconds for all keys (None = no expiration).

        Returns:
            Result with None on success, or CacheError.

        Example:
            result = await cache.set_many(
                {
                    "user:1": json.dumps(user1),
                    "user:2": json.dumps(user2),
                },
                ttl=3600
            )
            match result:
                case Success(_):
                    # All keys cached
                    pass
                case Failure(_):
                    # Fail open
                    pass
        """
        ...
Functions
get async 
get(key: str) -> Result[str | None, DomainError]

Get value from cache.

Parameters:

Name Type Description Default
key str

Cache key.

 required

Returns:

Type Description
Result[str | None, DomainError]

Result with value if found, None if not found, or CacheError.
Example

result = await cache.get("user:123") match result: case Success(value) if value: # Key found user_data = json.loads(value) case Success(None): # Key not found (cache miss) pass case Failure(error): # Cache error - fail open logger.warning("Cache get failed", error=error)

Source code in src/domain/protocols/cache_protocol.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def get(self, key: str) -> Result[str | None, DomainError]:
    """Get value from cache.

    Args:
        key: Cache key.

    Returns:
        Result with value if found, None if not found, or CacheError.

    Example:
        result = await cache.get("user:123")
        match result:
            case Success(value) if value:
                # Key found
                user_data = json.loads(value)
            case Success(None):
                # Key not found (cache miss)
                pass
            case Failure(error):
                # Cache error - fail open
                logger.warning("Cache get failed", error=error)
    """
    ...
get_json async 
get_json(
    key: str,
) -> Result[dict[str, Any] | None, DomainError]

Get JSON value from cache.

Convenience method that deserializes JSON automatically.

Parameters:

Name Type Description Default
key str

Cache key.

 required

Returns:

Type Description
Result[dict[str, Any] | None, DomainError]

Result with parsed dict if found, None if not found, or CacheError.
Example

result = await cache.get_json("user:123") match result: case Success(data) if data: user_id = data["id"] case Success(None): # Cache miss pass case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def get_json(self, key: str) -> Result[dict[str, Any] | None, DomainError]:
    """Get JSON value from cache.

    Convenience method that deserializes JSON automatically.

    Args:
        key: Cache key.

    Returns:
        Result with parsed dict if found, None if not found, or CacheError.

    Example:
        result = await cache.get_json("user:123")
        match result:
            case Success(data) if data:
                user_id = data["id"]
            case Success(None):
                # Cache miss
                pass
            case Failure(_):
                # Fail open
                pass
    """
    ...
set async 
set(
    key: str, value: str, ttl: int | None = None
) -> Result[None, DomainError]

Set value in cache.

Parameters:

Name Type Description Default
key str

Cache key.

 required
value str

Value to cache (string).

 required
ttl int | None

Time to live in seconds (None = no expiration).

 None

Returns:

Type Description
Result[None, DomainError]

Result with None on success, or CacheError.
Example

result = await cache.set( "user:123", json.dumps(user_data), ttl=3600 # 1 hour ) match result: case Success(): # Cached successfully pass case Failure(): # Fail open - continue without cache pass

Source code in src/domain/protocols/cache_protocol.py 
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
async def set(
    self,
    key: str,
    value: str,
    ttl: int | None = None,
) -> Result[None, DomainError]:
    """Set value in cache.

    Args:
        key: Cache key.
        value: Value to cache (string).
        ttl: Time to live in seconds (None = no expiration).

    Returns:
        Result with None on success, or CacheError.

    Example:
        result = await cache.set(
            "user:123",
            json.dumps(user_data),
            ttl=3600  # 1 hour
        )
        match result:
            case Success(_):
                # Cached successfully
                pass
            case Failure(_):
                # Fail open - continue without cache
                pass
    """
    ...
set_json async 
set_json(
    key: str, value: dict[str, Any], ttl: int | None = None
) -> Result[None, DomainError]

Set JSON value in cache.

Convenience method that serializes dict to JSON automatically.

Parameters:

Name Type Description Default
key str

Cache key.

 required
value dict[str, Any]

Dict to cache (will be JSON serialized).

 required
ttl int | None

Time to live in seconds (None = no expiration).

 None

Returns:

Type Description
Result[None, DomainError]

Result with None on success, or CacheError.
Example

result = await cache.set_json( "user:123", {"id": "123", "email": "user@example.com"}, ttl=1800 # 30 minutes )

Source code in src/domain/protocols/cache_protocol.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def set_json(
    self,
    key: str,
    value: dict[str, Any],
    ttl: int | None = None,
) -> Result[None, DomainError]:
    """Set JSON value in cache.

    Convenience method that serializes dict to JSON automatically.

    Args:
        key: Cache key.
        value: Dict to cache (will be JSON serialized).
        ttl: Time to live in seconds (None = no expiration).

    Returns:
        Result with None on success, or CacheError.

    Example:
        result = await cache.set_json(
            "user:123",
            {"id": "123", "email": "user@example.com"},
            ttl=1800  # 30 minutes
        )
    """
    ...
delete async 
delete(key: str) -> Result[bool, DomainError]

Delete key from cache.

Parameters:

Name Type Description Default
key str

Cache key to delete.

 required

Returns:

Type Description
Result[bool, DomainError]

Result with True if key was deleted, False if key didn't exist,
Result[bool, DomainError]

or CacheError.
Example

result = await cache.delete("user:123") match result: case Success(True): # Key was deleted pass case Success(False): # Key didn't exist pass case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
async def delete(self, key: str) -> Result[bool, DomainError]:
    """Delete key from cache.

    Args:
        key: Cache key to delete.

    Returns:
        Result with True if key was deleted, False if key didn't exist,
        or CacheError.

    Example:
        result = await cache.delete("user:123")
        match result:
            case Success(True):
                # Key was deleted
                pass
            case Success(False):
                # Key didn't exist
                pass
            case Failure(_):
                # Fail open
                pass
    """
    ...
exists async 
exists(key: str) -> Result[bool, DomainError]

Check if key exists in cache.

Parameters:

Name Type Description Default
key str

Cache key to check.

 required

Returns:

Type Description
Result[bool, DomainError]

Result with True if key exists, False if not, or CacheError.
Example

result = await cache.exists("user:123") match result: case Success(True): # Key exists pass case Success(False): # Key doesn't exist pass case Failure(_): # Fail open - assume doesn't exist pass

Source code in src/domain/protocols/cache_protocol.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
async def exists(self, key: str) -> Result[bool, DomainError]:
    """Check if key exists in cache.

    Args:
        key: Cache key to check.

    Returns:
        Result with True if key exists, False if not, or CacheError.

    Example:
        result = await cache.exists("user:123")
        match result:
            case Success(True):
                # Key exists
                pass
            case Success(False):
                # Key doesn't exist
                pass
            case Failure(_):
                # Fail open - assume doesn't exist
                pass
    """
    ...
expire async 
expire(key: str, seconds: int) -> Result[bool, DomainError]

Set expiration time on key.

Parameters:

Name Type Description Default
key str

Cache key.

 required
seconds int

Seconds until expiration.

 required

Returns:

Type Description
Result[bool, DomainError]

Result with True if timeout was set, False if key doesn't exist,
Result[bool, DomainError]

or CacheError.
Example

result = await cache.expire("session:abc", 1800) match result: case Success(True): # Expiration set pass case Success(False): # Key doesn't exist pass case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
async def expire(self, key: str, seconds: int) -> Result[bool, DomainError]:
    """Set expiration time on key.

    Args:
        key: Cache key.
        seconds: Seconds until expiration.

    Returns:
        Result with True if timeout was set, False if key doesn't exist,
        or CacheError.

    Example:
        result = await cache.expire("session:abc", 1800)
        match result:
            case Success(True):
                # Expiration set
                pass
            case Success(False):
                # Key doesn't exist
                pass
            case Failure(_):
                # Fail open
                pass
    """
    ...
ttl async 
ttl(key: str) -> Result[int | None, DomainError]

Get time to live for key.

Parameters:

Name Type Description Default
key str

Cache key.

 required

Returns:

Type Description
Result[int | None, DomainError]

Result with seconds until expiration, None if no TTL or key doesn't
Result[int | None, DomainError]

exist, or CacheError.
Example

result = await cache.ttl("session:abc") match result: case Success(seconds) if seconds: # Key expires in seconds pass case Success(None): # No TTL or key doesn't exist pass case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def ttl(self, key: str) -> Result[int | None, DomainError]:
    """Get time to live for key.

    Args:
        key: Cache key.

    Returns:
        Result with seconds until expiration, None if no TTL or key doesn't
        exist, or CacheError.

    Example:
        result = await cache.ttl("session:abc")
        match result:
            case Success(seconds) if seconds:
                # Key expires in `seconds`
                pass
            case Success(None):
                # No TTL or key doesn't exist
                pass
            case Failure(_):
                # Fail open
                pass
    """
    ...
increment async 
increment(
    key: str, amount: int = 1
) -> Result[int, DomainError]

Increment numeric value in cache (atomic).

If key doesn't exist, it's created with value = amount.

Parameters:

Name Type Description Default
key str

Cache key.

 required
amount int

Amount to increment by (default: 1).

 1

Returns:

Type Description
Result[int, DomainError]

Result with new value after increment, or CacheError.

Example (rate limiting): result = await cache.increment(f"rate_limit:{user_id}:{endpoint}") match result: case Success(count) if count == 1: # First request - set expiration await cache.expire(key, 60) case Success(count) if count > 100: # Rate limit exceeded raise RateLimitError() case Failure(_): # Fail open - allow request pass

Source code in src/domain/protocols/cache_protocol.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
async def increment(self, key: str, amount: int = 1) -> Result[int, DomainError]:
    """Increment numeric value in cache (atomic).

    If key doesn't exist, it's created with value = amount.

    Args:
        key: Cache key.
        amount: Amount to increment by (default: 1).

    Returns:
        Result with new value after increment, or CacheError.

    Example (rate limiting):
        result = await cache.increment(f"rate_limit:{user_id}:{endpoint}")
        match result:
            case Success(count) if count == 1:
                # First request - set expiration
                await cache.expire(key, 60)
            case Success(count) if count > 100:
                # Rate limit exceeded
                raise RateLimitError()
            case Failure(_):
                # Fail open - allow request
                pass
    """
    ...
decrement async 
decrement(
    key: str, amount: int = 1
) -> Result[int, DomainError]

Decrement numeric value in cache (atomic).

Parameters:

Name Type Description Default
key str

Cache key.

 required
amount int

Amount to decrement by (default: 1).

 1

Returns:

Type Description
Result[int, DomainError]

Result with new value after decrement, or CacheError.
Source code in src/domain/protocols/cache_protocol.py 
286
287
288
289
290
291
292
293
294
295
296
async def decrement(self, key: str, amount: int = 1) -> Result[int, DomainError]:
    """Decrement numeric value in cache (atomic).

    Args:
        key: Cache key.
        amount: Amount to decrement by (default: 1).

    Returns:
        Result with new value after decrement, or CacheError.
    """
    ...
flush async 
flush() -> Result[None, DomainError]

Flush all keys from cache.

WARNING: Use with extreme caution! This clears ALL cache data. Should only be used in tests or maintenance windows.

Returns:

Type Description
Result[None, DomainError]

Result with None on success, or CacheError.

Example (tests only): await cache.flush() # Clear cache between tests

Source code in src/domain/protocols/cache_protocol.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
async def flush(self) -> Result[None, DomainError]:
    """Flush all keys from cache.

    WARNING: Use with extreme caution! This clears ALL cache data.
    Should only be used in tests or maintenance windows.

    Returns:
        Result with None on success, or CacheError.

    Example (tests only):
        await cache.flush()  # Clear cache between tests
    """
    ...
ping async 
ping() -> Result[bool, DomainError]

Check cache connectivity (health check).

Returns:

Type Description
Result[bool, DomainError]

Result with True if cache is reachable, or CacheError.
Example

result = await cache.ping() match result: case Success(True): # Cache is healthy pass case Failure(_): # Cache unreachable logger.error("Cache health check failed")

Source code in src/domain/protocols/cache_protocol.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
async def ping(self) -> Result[bool, DomainError]:
    """Check cache connectivity (health check).

    Returns:
        Result with True if cache is reachable, or CacheError.

    Example:
        result = await cache.ping()
        match result:
            case Success(True):
                # Cache is healthy
                pass
            case Failure(_):
                # Cache unreachable
                logger.error("Cache health check failed")
    """
    ...
delete_pattern async 
delete_pattern(pattern: str) -> Result[int, DomainError]

Delete all keys matching pattern.

Parameters:

Name Type Description Default
pattern str

Glob-style pattern (e.g., "authz:user123:*").

 required

Returns:

Type Description
Result[int, DomainError]

Result with number of keys deleted, or CacheError.
Example

result = await cache.delete_pattern("session:user123:*") match result: case Success(count): logger.info(f"Deleted {count} keys") case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
async def delete_pattern(self, pattern: str) -> Result[int, DomainError]:
    """Delete all keys matching pattern.

    Args:
        pattern: Glob-style pattern (e.g., "authz:user123:*").

    Returns:
        Result with number of keys deleted, or CacheError.

    Example:
        result = await cache.delete_pattern("session:user123:*")
        match result:
            case Success(count):
                logger.info(f"Deleted {count} keys")
            case Failure(_):
                # Fail open
                pass
    """
    ...
get_many async 
get_many(
    keys: list[str],
) -> Result[dict[str, str | None], DomainError]

Get multiple values at once (batch operation).

More efficient than multiple get() calls - uses Redis MGET.

Parameters:

Name Type Description Default
keys list[str]

List of cache keys to retrieve.

 required

Returns:

Type Description
Result[dict[str, str | None], DomainError]

Result with dict mapping keys to values (None for missing keys),
Result[dict[str, str | None], DomainError]

or CacheError.
Example

result = await cache.get_many(["user:1", "user:2", "user:3"]) match result: case Success(data): for key, value in data.items(): if value is not None: # Process cached value pass case Failure(_): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
async def get_many(
    self, keys: list[str]
) -> Result[dict[str, str | None], DomainError]:
    """Get multiple values at once (batch operation).

    More efficient than multiple get() calls - uses Redis MGET.

    Args:
        keys: List of cache keys to retrieve.

    Returns:
        Result with dict mapping keys to values (None for missing keys),
        or CacheError.

    Example:
        result = await cache.get_many(["user:1", "user:2", "user:3"])
        match result:
            case Success(data):
                for key, value in data.items():
                    if value is not None:
                        # Process cached value
                        pass
            case Failure(_):
                # Fail open
                pass
    """
    ...
set_many async 
set_many(
    mapping: dict[str, str], ttl: int | None = None
) -> Result[None, DomainError]

Set multiple values at once (batch operation).

More efficient than multiple set() calls - uses Redis pipeline.

Parameters:

Name Type Description Default
mapping dict[str, str]

Dict of key->value pairs to cache.

 required
ttl int | None

Time to live in seconds for all keys (None = no expiration).

 None

Returns:

Type Description
Result[None, DomainError]

Result with None on success, or CacheError.
Example

result = await cache.set_many( { "user:1": json.dumps(user1), "user:2": json.dumps(user2), }, ttl=3600 ) match result: case Success(): # All keys cached pass case Failure(): # Fail open pass

Source code in src/domain/protocols/cache_protocol.py 
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
async def set_many(
    self,
    mapping: dict[str, str],
    ttl: int | None = None,
) -> Result[None, DomainError]:
    """Set multiple values at once (batch operation).

    More efficient than multiple set() calls - uses Redis pipeline.

    Args:
        mapping: Dict of key->value pairs to cache.
        ttl: Time to live in seconds for all keys (None = no expiration).

    Returns:
        Result with None on success, or CacheError.

    Example:
        result = await cache.set_many(
            {
                "user:1": json.dumps(user1),
                "user:2": json.dumps(user2),
            },
            ttl=3600
        )
        match result:
            case Success(_):
                # All keys cached
                pass
            case Failure(_):
                # Fail open
                pass
    """
    ...