Skip to content

Python API

rtc.RedisTaggedCache dataclass

A Redis-based caching system with tag-based invalidation support.

This class provides a thread-safe caching implementation using Redis as the backend. It supports tag-based cache invalidation, allowing you to invalidate multiple cache entries at once by invalidating their associated tags.

Features: - Thread-safe operations - Tag-based cache invalidation - Configurable cache entry lifetimes - Local memory cache option for testing - Automatic and customizable serialization support - Decorator interface for easy function result caching - Cache hooks for monitoring and debugging

Example 
cache = RedisTaggedCache(
    host="localhost",
    port=6379
)

# Basic usage
cache.set("key", "value", tags=["tag1", "tag2"])
assert cache.get("key") == "value"
cache.invalidate("tag1")  # Invalidates all entries with tag1

cache.get("key")  # it will raise a CacheMiss exception
Note

All operations are thread-safe and can be used in multi-threaded environments.

Source code in rtc/infra/controllers/lib.py 
 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
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
@dataclass
class RedisTaggedCache:
    """A Redis-based caching system with tag-based invalidation support.

    This class provides a thread-safe caching implementation using Redis as the backend.
    It supports tag-based cache invalidation, allowing you to invalidate multiple cache
    entries at once by invalidating their associated tags.

    Features:
    - Thread-safe operations
    - Tag-based cache invalidation
    - Configurable cache entry lifetimes
    - Local memory cache option for testing
    - Automatic and customizable serialization support
    - Decorator interface for easy function result caching
    - Cache hooks for monitoring and debugging

    Example:
        ```python
        cache = RedisTaggedCache(
            host="localhost",
            port=6379
        )

        # Basic usage
        cache.set("key", "value", tags=["tag1", "tag2"])
        assert cache.get("key") == "value"
        cache.invalidate("tag1")  # Invalidates all entries with tag1

        cache.get("key")  # it will raise a CacheMiss exception
        ```

    Note:
        All operations are thread-safe and can be used in multi-threaded environments.
    """

    namespace: str = "default"
    """Namespace prefix for all cache entries to avoid key collisions."""

    host: str = "localhost"
    """Redis server hostname or IP address."""

    port: int = 6379
    """Redis server port number."""

    db: int = 0
    """Redis database number (0-15)."""

    ssl: bool = False
    """Whether to use SSL/TLS for Redis connection."""

    socket_timeout: int = 5
    """Socket timeout in seconds for Redis operations."""

    socket_connect_timeout: int = 5
    """Socket connection timeout in seconds when establishing Redis connection."""

    default_lifetime: Optional[int] = 3600  # 1h
    """Default lifetime for cache entries in seconds.

    If set to None, entries will not expire automatically. In this case, ensure your
    Redis instance is configured with an appropriate eviction policy.
    """

    lifetime_for_tags: Optional[int] = 86400  # 24h
    """Lifetime for tag entries in seconds.

    When a tag expires or is invalidated, all cache entries associated with that tag
    are also invalidated. Set to None for no automatic tag expiration.
    """

    disabled: bool = False
    """If True, disables all caching operations while maintaining the API interface.

    Useful for testing or temporarily disabling cache without code changes.
    """

    in_local_memory: bool = False
    """If True, uses process-local memory instead of Redis for storage.

    Warning:
        This mode is intended for testing only and should not be used in production
        as it doesn't provide cross-process cache consistency.
    """

    cache_hook: Optional[CacheHook] = None
    """Optional callback function for monitoring cache operations.

    The hook function is called after each cache decorator operation with the following signature:
    ```python
    def cache_hook(
        key: str,
        tags: Iterable[str],
        cache_info: CacheInfo,
        userdata: Optional[Any] = None
    ) -> None:
        pass
    ```

    Parameters:
        - key: The cache key being accessed
        - tags: Iterable of tags associated with the key
        - cache_info: Object containing cache operation metrics
        - userdata: Optional custom data passed through the decorator
    """

    serializer: Callable[[Any], Optional[bytes]] = DEFAULT_SERIALIZER
    """Function to serialize Python objects before storing in cache.

    Must accept any Python object and return bytes or None (means: cache bypassed).
    """

    unserializer: Callable[[bytes], Any] = DEFAULT_UNSERIALIZER
    """Function to deserialize data read from cache back into Python objects.

    Must accept bytes and return the original Python object.
    """

    _internal_lock: Lock = field(init=False, default_factory=Lock)
    _forced_metadata_adapter: Optional[MetadataPort] = field(
        init=False, default=None
    )  # for advanced usage only
    _forced_storage_adapter: Optional[StoragePort] = field(
        init=False, default=None
    )  # for advanced usage only
    __service: Optional[Service] = field(
        init=False, default=None
    )  # cache of the Service object

    @property
    def _service(self) -> Service:
        with self._internal_lock:
            if self.__service is None:
                self.__service = self._make_service()
            return self.__service

    def _make_service(self) -> Service:
        metadata_adapter: MetadataPort
        storage_adapter: StoragePort
        redis_kwargs = {
            "host": self.host,
            "port": self.port,
            "db": self.db,
            "ssl": self.ssl,
            "socket_timeout": self.socket_timeout,
            "socket_connect_timeout": self.socket_connect_timeout,
        }
        if self._forced_metadata_adapter:
            metadata_adapter = self._forced_metadata_adapter
        elif self.disabled:
            metadata_adapter = BlackHoleMetadataAdapter()
        elif self.in_local_memory:
            metadata_adapter = DictMetadataAdapter()
        else:
            metadata_adapter = RedisMetadataAdapter(redis_kwargs)
        if self._forced_storage_adapter:
            storage_adapter = self._forced_storage_adapter
        elif self.disabled:
            storage_adapter = BlackHoleStorageAdapter()
        elif self.in_local_memory:
            storage_adapter = DictStorageAdapter()
        else:
            storage_adapter = RedisStorageAdapter(redis_kwargs)
        return Service(
            namespace=self.namespace,
            metadata_service=MetadataService(
                namespace=self.namespace,
                adapter=metadata_adapter,
                default_lifetime=self.lifetime_for_tags or 0,
            ),
            storage_service=StorageService(
                namespace=self.namespace,
                adapter=storage_adapter,
                default_lifetime=self.default_lifetime or 0,
            ),
            cache_hook=self.cache_hook,
            serializer=self.serializer,
            unserializer=self.unserializer,
        )

    def _rebuild_service(self):
        with self._internal_lock:
            self.__service = None

    def set(
        self,
        key: str,
        value: Any,
        tags: Optional[Iterable[str]] = None,
        lifetime: Optional[int] = None,
    ) -> bool:
        """Store a value in the cache with optional tags and lifetime.

        Args:
            key: Unique identifier for the cache entry
            value: Any Python object to store (must be serializable)
            tags: Optional list of tags for invalidation
            lifetime: Optional TTL in seconds (if set: overrides default_lifetime, 0 means no expiration)

        Returns:
            bool: True if the value was successfully stored, False otherwise

        Example:
            ```python
            cache.set("user:123", user_data, tags=["user", "user:123"], lifetime=3600)
            ```
        """
        return self._service.set(key, value, tags, lifetime)

    def delete(self, key: str, tags: Optional[Iterable[str]] = None) -> bool:
        """Remove an entry from the cache.

        Args:
            key: The key to delete
            tags: Optional list of tags (for consistency, should match set() tags)

        Returns:
            bool: True if the key was found and deleted, False if it didn't exist

        Note:
            No exception is raised if the key doesn't exist or was already invalidated.
        """
        return self._service.delete(key, tags)

    def get(
        self,
        key: str,
        tags: Optional[Iterable[str]] = None,
    ) -> Any:
        """Retrieve a value from the cache.

        Args:
            key: The key to look up
            tags: Optional list of tags (for consistency, should match set() tags)

        Returns:
            The stored value if found and valid

        Raises:
            CacheMiss: If the key doesn't exist, has expired, or was invalidated

        Example:
            ```python
            try:
                value = cache.get("user:123", tags=["user", "user:123"])
            except CacheMiss:
                value = compute_user_data()
            ```
        """
        return self._service.get(key, tags)

    def invalidate(self, tags: Union[str, Iterable[str]]) -> bool:
        """Invalidate all cache entries associated with the given tag(s).

        Args:
            tags: Single tag string or iterable of tags to invalidate

        Returns:
            bool: True if the invalidation was successful

        Example:
            ```python
            # Invalidate all user-related cache entries
            cache.invalidate("user")

            # Invalidate multiple tags at once
            cache.invalidate(["user:123", "session:456"])
            ```
        """
        if isinstance(tags, str):
            return self._service.invalidate_tags([tags])
        else:
            return self._service.invalidate_tags(tags)

    def invalidate_all(self) -> bool:
        """Invalidate all cache entries in the current namespace.

        This operation is implemented efficiently using a special tag that is
        automatically associated with all cache entries. The operation is O(1)
        regardless of the number of cache entries.

        Returns:
            bool: True if the invalidation was successful

        Example:
            ```python
            # Clear entire cache (within current namespace)
            cache.invalidate_all()
            ```
        """
        return self._service.invalidate_all()

    def decorator(
        self,
        tags: Optional[Union[Iterable[str], Callable[..., Iterable[str]]]] = None,
        lifetime: Optional[int] = None,
        key: Optional[Callable[..., str]] = None,
        hook_userdata: Optional[Any] = None,
        lock: bool = False,
        lock_timeout: int = 5,
        serializer: Optional[Callable[[Any], Optional[bytes]]] = None,
        unserializer: Optional[Callable[[bytes], Any]] = None,
    ) -> Callable:
        """Decorator for automatically caching function results.

        This decorator provides a high-level interface for caching function return values.
        It supports both static and dynamic cache keys and tags, custom serialization,
        and protection against cache stampede.

        If you don't provide a key, we will generate a key from the function name and arguments
        (that must be JSON-serializable). For methods, we will use the method name and class name
        but the instance (self) won't be taken into account by default (for generating the key).

        Args:
            tags: Static list of tags or a function that generates tags dynamically
            lifetime: Optional TTL in seconds (overrides default_lifetime)
            key: Optional function to generate custom cache keys
            hook_userdata: Optional data passed to cache hooks
            lock: If True, uses distributed locking to prevent cache stampede
            lock_timeout: Lock timeout in seconds (default: 5)
            serializer: Optional custom serializer for this function
            unserializer: Optional custom unserializer for this function

        Returns:
            A decorator function that can be applied to methods or functions

        Example:
            ```python
            # Basic usage with static tags
            @cache.decorator(tags=["user"])
            def get_user(user_id: int) -> dict:
                return db.fetch_user(user_id)

            # Dynamic tags based on function arguments
            @cache.decorator(
                tags=lambda user_id: [f"user:{user_id}", "user"],
                lifetime=3600
            )
            def get_user_with_dynamic_tags(user_id: int) -> dict:
                return db.fetch_user(user_id)

            # Custom cache key generation
            @cache.decorator(
                key=lambda user_id: f"user_profile:{user_id}"
            )
            def get_user_profile(user_id: int) -> dict:
                return db.fetch_user_profile(user_id)

            # Protection against cache stampede
            @cache.decorator(lock=True, lock_timeout=10)
            def expensive_computation() -> dict:
                return perform_slow_calculation()
            ```

        Note:
            - The decorated function's arguments must be JSON-serializable for automatic key generation
            - When using custom key functions, ensure keys are unique and deterministic
            - Lock timeout should be greater than the expected function execution time
        """
        return cache_decorator(
            service=self._service,
            serializer=serializer if serializer else self.serializer,
            unserializer=unserializer if unserializer else self.unserializer,
            lock=lock,
            lock_timeout=lock_timeout,
            key=key,
            hook_userdata=hook_userdata,
            tags=tags,
            lifetime=lifetime,
        )

    def function_decorator(self, *args, **kwargs):
        """DEPRECATED: Use `decorator()` instead.

        This method will be removed in a future version.
        """
        return self.decorator(*args, **kwargs)

    def method_decorator(self, *args, **kwargs):
        """DEPRECATED: Use `decorator()` instead.

        This method will be removed in a future version.
        """
        return self.decorator(*args, **kwargs)

cache_hook: Optional[CacheHook] = None class-attribute instance-attribute

Optional callback function for monitoring cache operations.

The hook function is called after each cache decorator operation with the following signature: 

def cache_hook(
    key: str,
    tags: Iterable[str],
    cache_info: CacheInfo,
    userdata: Optional[Any] = None
) -> None:
    pass

Parameters:

Name Type Description Default
- key

The cache key being accessed

 required
- tags

Iterable of tags associated with the key

 required
- cache_info

Object containing cache operation metrics

 required
- userdata

Optional custom data passed through the decorator

 required

db: int = 0 class-attribute instance-attribute

Redis database number (0-15).

default_lifetime: Optional[int] = 3600 class-attribute instance-attribute

Default lifetime for cache entries in seconds.

If set to None, entries will not expire automatically. In this case, ensure your Redis instance is configured with an appropriate eviction policy.

disabled: bool = False class-attribute instance-attribute

If True, disables all caching operations while maintaining the API interface.

Useful for testing or temporarily disabling cache without code changes.

host: str = 'localhost' class-attribute instance-attribute

Redis server hostname or IP address.

in_local_memory: bool = False class-attribute instance-attribute

If True, uses process-local memory instead of Redis for storage.

Warning

This mode is intended for testing only and should not be used in production as it doesn't provide cross-process cache consistency.

lifetime_for_tags: Optional[int] = 86400 class-attribute instance-attribute

Lifetime for tag entries in seconds.

When a tag expires or is invalidated, all cache entries associated with that tag are also invalidated. Set to None for no automatic tag expiration.

namespace: str = 'default' class-attribute instance-attribute

Namespace prefix for all cache entries to avoid key collisions.

port: int = 6379 class-attribute instance-attribute

Redis server port number.

serializer: Callable[[Any], Optional[bytes]] = DEFAULT_SERIALIZER class-attribute instance-attribute

Function to serialize Python objects before storing in cache.

Must accept any Python object and return bytes or None (means: cache bypassed).

socket_connect_timeout: int = 5 class-attribute instance-attribute

Socket connection timeout in seconds when establishing Redis connection.

socket_timeout: int = 5 class-attribute instance-attribute

Socket timeout in seconds for Redis operations.

ssl: bool = False class-attribute instance-attribute

Whether to use SSL/TLS for Redis connection.

unserializer: Callable[[bytes], Any] = DEFAULT_UNSERIALIZER class-attribute instance-attribute

Function to deserialize data read from cache back into Python objects.

Must accept bytes and return the original Python object.

decorator(tags: Optional[Union[Iterable[str], Callable[..., Iterable[str]]]] = None, lifetime: Optional[int] = None, key: Optional[Callable[..., str]] = None, hook_userdata: Optional[Any] = None, lock: bool = False, lock_timeout: int = 5, serializer: Optional[Callable[[Any], Optional[bytes]]] = None, unserializer: Optional[Callable[[bytes], Any]] = None) -> Callable

Decorator for automatically caching function results.

This decorator provides a high-level interface for caching function return values. It supports both static and dynamic cache keys and tags, custom serialization, and protection against cache stampede.

If you don't provide a key, we will generate a key from the function name and arguments (that must be JSON-serializable). For methods, we will use the method name and class name but the instance (self) won't be taken into account by default (for generating the key).

Parameters:

Name Type Description Default
tags Optional[Union[Iterable[str], Callable[..., Iterable[str]]]]

Static list of tags or a function that generates tags dynamically

 None
lifetime Optional[int]

Optional TTL in seconds (overrides default_lifetime)

 None
key Optional[Callable[..., str]]

Optional function to generate custom cache keys

 None
hook_userdata Optional[Any]

Optional data passed to cache hooks

 None
lock bool

If True, uses distributed locking to prevent cache stampede

 False
lock_timeout int

Lock timeout in seconds (default: 5)

 5
serializer Optional[Callable[[Any], Optional[bytes]]]

Optional custom serializer for this function

 None
unserializer Optional[Callable[[bytes], Any]]

Optional custom unserializer for this function

 None

Returns:

Type Description
Callable

A decorator function that can be applied to methods or functions
Example 
# Basic usage with static tags
@cache.decorator(tags=["user"])
def get_user(user_id: int) -> dict:
    return db.fetch_user(user_id)

# Dynamic tags based on function arguments
@cache.decorator(
    tags=lambda user_id: [f"user:{user_id}", "user"],
    lifetime=3600
)
def get_user_with_dynamic_tags(user_id: int) -> dict:
    return db.fetch_user(user_id)

# Custom cache key generation
@cache.decorator(
    key=lambda user_id: f"user_profile:{user_id}"
)
def get_user_profile(user_id: int) -> dict:
    return db.fetch_user_profile(user_id)

# Protection against cache stampede
@cache.decorator(lock=True, lock_timeout=10)
def expensive_computation() -> dict:
    return perform_slow_calculation()
Note
  • The decorated function's arguments must be JSON-serializable for automatic key generation
  • When using custom key functions, ensure keys are unique and deterministic
  • Lock timeout should be greater than the expected function execution time
Source code in rtc/infra/controllers/lib.py 
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
def decorator(
    self,
    tags: Optional[Union[Iterable[str], Callable[..., Iterable[str]]]] = None,
    lifetime: Optional[int] = None,
    key: Optional[Callable[..., str]] = None,
    hook_userdata: Optional[Any] = None,
    lock: bool = False,
    lock_timeout: int = 5,
    serializer: Optional[Callable[[Any], Optional[bytes]]] = None,
    unserializer: Optional[Callable[[bytes], Any]] = None,
) -> Callable:
    """Decorator for automatically caching function results.

    This decorator provides a high-level interface for caching function return values.
    It supports both static and dynamic cache keys and tags, custom serialization,
    and protection against cache stampede.

    If you don't provide a key, we will generate a key from the function name and arguments
    (that must be JSON-serializable). For methods, we will use the method name and class name
    but the instance (self) won't be taken into account by default (for generating the key).

    Args:
        tags: Static list of tags or a function that generates tags dynamically
        lifetime: Optional TTL in seconds (overrides default_lifetime)
        key: Optional function to generate custom cache keys
        hook_userdata: Optional data passed to cache hooks
        lock: If True, uses distributed locking to prevent cache stampede
        lock_timeout: Lock timeout in seconds (default: 5)
        serializer: Optional custom serializer for this function
        unserializer: Optional custom unserializer for this function

    Returns:
        A decorator function that can be applied to methods or functions

    Example:
        ```python
        # Basic usage with static tags
        @cache.decorator(tags=["user"])
        def get_user(user_id: int) -> dict:
            return db.fetch_user(user_id)

        # Dynamic tags based on function arguments
        @cache.decorator(
            tags=lambda user_id: [f"user:{user_id}", "user"],
            lifetime=3600
        )
        def get_user_with_dynamic_tags(user_id: int) -> dict:
            return db.fetch_user(user_id)

        # Custom cache key generation
        @cache.decorator(
            key=lambda user_id: f"user_profile:{user_id}"
        )
        def get_user_profile(user_id: int) -> dict:
            return db.fetch_user_profile(user_id)

        # Protection against cache stampede
        @cache.decorator(lock=True, lock_timeout=10)
        def expensive_computation() -> dict:
            return perform_slow_calculation()
        ```

    Note:
        - The decorated function's arguments must be JSON-serializable for automatic key generation
        - When using custom key functions, ensure keys are unique and deterministic
        - Lock timeout should be greater than the expected function execution time
    """
    return cache_decorator(
        service=self._service,
        serializer=serializer if serializer else self.serializer,
        unserializer=unserializer if unserializer else self.unserializer,
        lock=lock,
        lock_timeout=lock_timeout,
        key=key,
        hook_userdata=hook_userdata,
        tags=tags,
        lifetime=lifetime,
    )

delete(key: str, tags: Optional[Iterable[str]] = None) -> bool

Remove an entry from the cache.

Parameters:

Name Type Description Default
key str

The key to delete

 required
tags Optional[Iterable[str]]

Optional list of tags (for consistency, should match set() tags)

 None

Returns:

Name Type Description
bool bool

True if the key was found and deleted, False if it didn't exist
Note

No exception is raised if the key doesn't exist or was already invalidated.

Source code in rtc/infra/controllers/lib.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def delete(self, key: str, tags: Optional[Iterable[str]] = None) -> bool:
    """Remove an entry from the cache.

    Args:
        key: The key to delete
        tags: Optional list of tags (for consistency, should match set() tags)

    Returns:
        bool: True if the key was found and deleted, False if it didn't exist

    Note:
        No exception is raised if the key doesn't exist or was already invalidated.
    """
    return self._service.delete(key, tags)

function_decorator(*args, **kwargs)

DEPRECATED: Use decorator() instead.

This method will be removed in a future version.

Source code in rtc/infra/controllers/lib.py 
392
393
394
395
396
397
def function_decorator(self, *args, **kwargs):
    """DEPRECATED: Use `decorator()` instead.

    This method will be removed in a future version.
    """
    return self.decorator(*args, **kwargs)

get(key: str, tags: Optional[Iterable[str]] = None) -> Any

Retrieve a value from the cache.

Parameters:

Name Type Description Default
key str

The key to look up

 required
tags Optional[Iterable[str]]

Optional list of tags (for consistency, should match set() tags)

 None

Returns:

Type Description
Any

The stored value if found and valid

Raises:

Type Description
CacheMiss

If the key doesn't exist, has expired, or was invalidated
Example 
try:
    value = cache.get("user:123", tags=["user", "user:123"])
except CacheMiss:
    value = compute_user_data()
Source code in rtc/infra/controllers/lib.py 
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
def get(
    self,
    key: str,
    tags: Optional[Iterable[str]] = None,
) -> Any:
    """Retrieve a value from the cache.

    Args:
        key: The key to look up
        tags: Optional list of tags (for consistency, should match set() tags)

    Returns:
        The stored value if found and valid

    Raises:
        CacheMiss: If the key doesn't exist, has expired, or was invalidated

    Example:
        ```python
        try:
            value = cache.get("user:123", tags=["user", "user:123"])
        except CacheMiss:
            value = compute_user_data()
        ```
    """
    return self._service.get(key, tags)

invalidate(tags: Union[str, Iterable[str]]) -> bool

Invalidate all cache entries associated with the given tag(s).

Parameters:

Name Type Description Default
tags Union[str, Iterable[str]]

Single tag string or iterable of tags to invalidate

 required

Returns:

Name Type Description
bool bool

True if the invalidation was successful
Example 
# Invalidate all user-related cache entries
cache.invalidate("user")

# Invalidate multiple tags at once
cache.invalidate(["user:123", "session:456"])
Source code in rtc/infra/controllers/lib.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def invalidate(self, tags: Union[str, Iterable[str]]) -> bool:
    """Invalidate all cache entries associated with the given tag(s).

    Args:
        tags: Single tag string or iterable of tags to invalidate

    Returns:
        bool: True if the invalidation was successful

    Example:
        ```python
        # Invalidate all user-related cache entries
        cache.invalidate("user")

        # Invalidate multiple tags at once
        cache.invalidate(["user:123", "session:456"])
        ```
    """
    if isinstance(tags, str):
        return self._service.invalidate_tags([tags])
    else:
        return self._service.invalidate_tags(tags)

invalidate_all() -> bool

Invalidate all cache entries in the current namespace.

This operation is implemented efficiently using a special tag that is automatically associated with all cache entries. The operation is O(1) regardless of the number of cache entries.

Returns:

Name Type Description
bool bool

True if the invalidation was successful
Example 
# Clear entire cache (within current namespace)
cache.invalidate_all()
Source code in rtc/infra/controllers/lib.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def invalidate_all(self) -> bool:
    """Invalidate all cache entries in the current namespace.

    This operation is implemented efficiently using a special tag that is
    automatically associated with all cache entries. The operation is O(1)
    regardless of the number of cache entries.

    Returns:
        bool: True if the invalidation was successful

    Example:
        ```python
        # Clear entire cache (within current namespace)
        cache.invalidate_all()
        ```
    """
    return self._service.invalidate_all()

method_decorator(*args, **kwargs)

DEPRECATED: Use decorator() instead.

This method will be removed in a future version.

Source code in rtc/infra/controllers/lib.py 
399
400
401
402
403
404
def method_decorator(self, *args, **kwargs):
    """DEPRECATED: Use `decorator()` instead.

    This method will be removed in a future version.
    """
    return self.decorator(*args, **kwargs)

set(key: str, value: Any, tags: Optional[Iterable[str]] = None, lifetime: Optional[int] = None) -> bool

Store a value in the cache with optional tags and lifetime.

Parameters:

Name Type Description Default
key str

Unique identifier for the cache entry

 required
value Any

Any Python object to store (must be serializable)

 required
tags Optional[Iterable[str]]

Optional list of tags for invalidation

 None
lifetime Optional[int]

Optional TTL in seconds (if set: overrides default_lifetime, 0 means no expiration)

 None

Returns:

Name Type Description
bool bool

True if the value was successfully stored, False otherwise
Example 
cache.set("user:123", user_data, tags=["user", "user:123"], lifetime=3600)
Source code in rtc/infra/controllers/lib.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def set(
    self,
    key: str,
    value: Any,
    tags: Optional[Iterable[str]] = None,
    lifetime: Optional[int] = None,
) -> bool:
    """Store a value in the cache with optional tags and lifetime.

    Args:
        key: Unique identifier for the cache entry
        value: Any Python object to store (must be serializable)
        tags: Optional list of tags for invalidation
        lifetime: Optional TTL in seconds (if set: overrides default_lifetime, 0 means no expiration)

    Returns:
        bool: True if the value was successfully stored, False otherwise

    Example:
        ```python
        cache.set("user:123", user_data, tags=["user", "user:123"], lifetime=3600)
        ```
    """
    return self._service.set(key, value, tags, lifetime)

rtc.CacheInfo dataclass

Class containing location infos about the cache call.

This is only used in cache hit/miss hooks.

Source code in rtc/app/types.py 
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
@dataclass
class CacheInfo:
    """Class containing location infos about the cache call.

    This is only used in cache hit/miss hooks.

    """

    filepath: str = ""
    """File path of the decorated function."""

    class_name: str = ""
    """Class name (empty for functions) of the decorated function."""

    function_name: str = ""
    """Function name of the decorated function/method."""

    function_args: Tuple[Any, ...] = field(default_factory=tuple)
    """Decorated function/method arguments (including self as first argument for methods) (*args)."""

    function_kwargs: Dict[str, Any] = field(default_factory=dict)
    """Decorated function/method keyword arguments (**kwargs)."""

    method_decorator: bool = False
    """If True, we decorated a method (and not a simple function)."""

    hit: bool = False
    """Cache hit (the value was found in the cache)."""

    elapsed: float = 0.0
    """Total elapsed time (in seconds). It includes the decorated function call in case of cache miss but excludes hooks."""

    decorated_elapsed: float = 0.0
    """Elapsed time of the decorated function call (in seconds), only in case of cache miss."""

    lock_waiting_ms: int = 0
    """Lock waiting time (in ms), only when used with cache decorators and lock=True."""

    lock_full_hit: bool = False
    """Lock full hit (no lock acquired at all, the value was cached before), only when used with cache decorators and lock=True."""

    lock_full_miss: bool = False
    """Lock full miss (we acquired a lock but the value was not cached after that => full cache miss), only when used with cache decorators and lock=True."""

    serialized_size: int = 0
    """Serialized size of the value (in bytes)."""

    # extra note: if lock_full_hit = False and lock_full_miss = False (when used with cache decorators and lock=True),
    # it means that the value was initially not here, so we acquired a lock but the value was cached after that (anti-dogpile effect)

    def _dump(self) -> List[str]:
        # Special method for cache decorators
        return [self.filepath, self.class_name, self.function_name]

class_name: str = '' class-attribute instance-attribute

Class name (empty for functions) of the decorated function.

decorated_elapsed: float = 0.0 class-attribute instance-attribute

Elapsed time of the decorated function call (in seconds), only in case of cache miss.

elapsed: float = 0.0 class-attribute instance-attribute

Total elapsed time (in seconds). It includes the decorated function call in case of cache miss but excludes hooks.

filepath: str = '' class-attribute instance-attribute

File path of the decorated function.

function_args: Tuple[Any, ...] = field(default_factory=tuple) class-attribute instance-attribute

Decorated function/method arguments (including self as first argument for methods) (*args).

function_kwargs: Dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute

Decorated function/method keyword arguments (**kwargs).

function_name: str = '' class-attribute instance-attribute

Function name of the decorated function/method.

hit: bool = False class-attribute instance-attribute

Cache hit (the value was found in the cache).

lock_full_hit: bool = False class-attribute instance-attribute

Lock full hit (no lock acquired at all, the value was cached before), only when used with cache decorators and lock=True.

lock_full_miss: bool = False class-attribute instance-attribute

Lock full miss (we acquired a lock but the value was not cached after that => full cache miss), only when used with cache decorators and lock=True.

lock_waiting_ms: int = 0 class-attribute instance-attribute

Lock waiting time (in ms), only when used with cache decorators and lock=True.

method_decorator: bool = False class-attribute instance-attribute

If True, we decorated a method (and not a simple function).

serialized_size: int = 0 class-attribute instance-attribute

Serialized size of the value (in bytes).

rtc.CacheMiss

Bases: Exception

Exception raised when a cache miss occurs.

Source code in rtc/app/exc.py 
1
2
3
4
class CacheMiss(Exception):
    """Exception raised when a cache miss occurs."""

    pass

rtc.CacheHook

Bases: Protocol

Source code in rtc/app/types.py 
70
71
72
73
74
75
76
77
78
79
class CacheHook(Protocol):
    def __call__(
        self,
        cache_key: str,
        cache_tags: Iterable[str],
        cache_info: CacheInfo,
        userdata: Any = None,
    ) -> None:
        """Signature of cache hooks."""
        pass

__call__(cache_key: str, cache_tags: Iterable[str], cache_info: CacheInfo, userdata: Any = None) -> None

Signature of cache hooks.

Source code in rtc/app/types.py 
71
72
73
74
75
76
77
78
79
def __call__(
    self,
    cache_key: str,
    cache_tags: Iterable[str],
    cache_info: CacheInfo,
    userdata: Any = None,
) -> None:
    """Signature of cache hooks."""
    pass