Skip to content

forge.cache

Decorator-based caching with in-memory LRU and Redis backends.

forge.cache

Classes

CacheBackendError

Bases: CacheError

Raised when a cache backend operation fails or is unavailable.

Source code in src/forge/cache/exceptions.py 
10
11
class CacheBackendError(CacheError):
    """Raised when a cache backend operation fails or is unavailable."""

CacheError

Bases: ForgeError

Base exception for all cache-related errors.

Source code in src/forge/cache/exceptions.py 
6
7
class CacheError(ForgeError):
    """Base exception for all cache-related errors."""

CacheModule

Bases: ForgeModule

Manages caching services and pluggable storage backends for the forge runtime.

Source code in src/forge/cache/module.py 
 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
139
140
141
142
143
144
145
class CacheModule(ForgeModule):
    """
    Manages caching services and pluggable storage backends for the forge runtime.
    """

    name = "cache"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._backend: CacheBackend | None = None
        self._runtime: Runtime | None = None

    @property
    def backend(self) -> CacheBackend:
        """Get the active cache backend."""
        if self._backend is None:
            raise CacheError("Cache backend is not initialized.")
        return self._backend

    async def setup(self, runtime: Runtime) -> None:
        """Initialize cache module and configure settings."""
        self._runtime = runtime
        set_cache_module(self)

        # Read cache configuration
        from forge.config.module import ConfigModule

        config_module = cast("ConfigModule", runtime.get(ConfigModule))
        config = getattr(config_module.config, "cache", None)

        backend_type = "memory"
        default_ttl = 300
        memory_max_size = 1000
        redis_url = None
        redis_key_prefix = "forge:"
        redis_max_connections = 10

        if config is not None:
            backend_type = getattr(config, "backend", "memory")
            default_ttl = getattr(config, "default_ttl", 300)
            memory_max_size = getattr(config, "memory_max_size", 1000)
            redis_config = getattr(config, "redis", None)
            if redis_config is not None:
                redis_url = getattr(redis_config, "url", None)
                redis_key_prefix = getattr(redis_config, "key_prefix", "forge:")
                redis_max_connections = getattr(redis_config, "max_connections", 10)

        if backend_type == "redis":
            from forge.cache.backends.redis import RedisBackend

            url = redis_url or "redis://localhost:6379/0"
            redis_backend = RedisBackend(
                redis_url=url,
                key_prefix=redis_key_prefix,
                max_connections=redis_max_connections,
                default_ttl=default_ttl,
            )
            await redis_backend.connect()
            self._backend = redis_backend
        else:
            from forge.cache.backends.memory import MemoryBackend

            self._backend = MemoryBackend(
                max_size=memory_max_size,
                default_ttl=default_ttl,
            )

    async def teardown(self) -> None:
        """Teardown the cache module."""
        set_cache_module(None)
        if self._backend:
            await self._backend.close()
            self._backend = None
        self._runtime = None

    async def get(self, key: str) -> Any | None:
        """Get a value from cache and emit hit/miss event."""
        if not self._backend:
            raise CacheError("Cache module is not initialized.")
        val = await self._backend.get(key)
        if self._runtime:
            if val is not None:
                await self._runtime._events.emit("cache.hit", key=key)
            else:
                await self._runtime._events.emit("cache.miss", key=key)
        return val

    async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
        """Set a cache value and emit set event."""
        if not self._backend:
            raise CacheError("Cache module is not initialized.")
        await self._backend.set(key, value, ttl)
        if self._runtime:
            await self._runtime._events.emit("cache.set", key=key, ttl=ttl)

    async def delete(self, key: str) -> bool:
        """Delete a key from cache."""
        if not self._backend:
            raise CacheError("Cache module is not initialized.")
        return await self._backend.delete(key)

    async def has(self, key: str) -> bool:
        """Check if a key exists in cache."""
        if not self._backend:
            raise CacheError("Cache module is not initialized.")
        return await self._backend.has(key)

    async def clear(self) -> None:
        """Clear all cache values."""
        if not self._backend:
            raise CacheError("Cache module is not initialized.")
        await self._backend.clear()

    def health_check(self) -> HealthResult:
        """Check the health status of the caching backend."""
        if self._backend is None:
            return HealthResult.error("Cache backend not initialized")

        from forge.cache.backends.memory import MemoryBackend

        if isinstance(self._backend, MemoryBackend):
            return HealthResult(HealthResult.OK, "Memory cache active")

        from forge.cache.backends.redis import RedisBackend

        if isinstance(self._backend, RedisBackend):
            from forge.core.redis_health import check_redis_health

            return check_redis_health(self._backend.url, label="Redis cache")

        return HealthResult.ok()
Attributes
backend property 
backend: CacheBackend

Get the active cache backend.

Methods:
clear async 
clear() -> None

Clear all cache values.

Source code in src/forge/cache/module.py 
122
123
124
125
126
async def clear(self) -> None:
    """Clear all cache values."""
    if not self._backend:
        raise CacheError("Cache module is not initialized.")
    await self._backend.clear()
delete async 
delete(key: str) -> bool

Delete a key from cache.

Source code in src/forge/cache/module.py 
110
111
112
113
114
async def delete(self, key: str) -> bool:
    """Delete a key from cache."""
    if not self._backend:
        raise CacheError("Cache module is not initialized.")
    return await self._backend.delete(key)
get async 
get(key: str) -> Any | None

Get a value from cache and emit hit/miss event.

Source code in src/forge/cache/module.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def get(self, key: str) -> Any | None:
    """Get a value from cache and emit hit/miss event."""
    if not self._backend:
        raise CacheError("Cache module is not initialized.")
    val = await self._backend.get(key)
    if self._runtime:
        if val is not None:
            await self._runtime._events.emit("cache.hit", key=key)
        else:
            await self._runtime._events.emit("cache.miss", key=key)
    return val
has async 
has(key: str) -> bool

Check if a key exists in cache.

Source code in src/forge/cache/module.py 
116
117
118
119
120
async def has(self, key: str) -> bool:
    """Check if a key exists in cache."""
    if not self._backend:
        raise CacheError("Cache module is not initialized.")
    return await self._backend.has(key)
health_check 
health_check() -> HealthResult

Check the health status of the caching backend.

Source code in src/forge/cache/module.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def health_check(self) -> HealthResult:
    """Check the health status of the caching backend."""
    if self._backend is None:
        return HealthResult.error("Cache backend not initialized")

    from forge.cache.backends.memory import MemoryBackend

    if isinstance(self._backend, MemoryBackend):
        return HealthResult(HealthResult.OK, "Memory cache active")

    from forge.cache.backends.redis import RedisBackend

    if isinstance(self._backend, RedisBackend):
        from forge.core.redis_health import check_redis_health

        return check_redis_health(self._backend.url, label="Redis cache")

    return HealthResult.ok()
set async 
set(key: str, value: Any, ttl: int | None = None) -> None

Set a cache value and emit set event.

Source code in src/forge/cache/module.py 
102
103
104
105
106
107
108
async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
    """Set a cache value and emit set event."""
    if not self._backend:
        raise CacheError("Cache module is not initialized.")
    await self._backend.set(key, value, ttl)
    if self._runtime:
        await self._runtime._events.emit("cache.set", key=key, ttl=ttl)
setup async 
setup(runtime: ForgeRuntime) -> None

Initialize cache module and configure settings.

Source code in src/forge/cache/module.py 
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
async def setup(self, runtime: Runtime) -> None:
    """Initialize cache module and configure settings."""
    self._runtime = runtime
    set_cache_module(self)

    # Read cache configuration
    from forge.config.module import ConfigModule

    config_module = cast("ConfigModule", runtime.get(ConfigModule))
    config = getattr(config_module.config, "cache", None)

    backend_type = "memory"
    default_ttl = 300
    memory_max_size = 1000
    redis_url = None
    redis_key_prefix = "forge:"
    redis_max_connections = 10

    if config is not None:
        backend_type = getattr(config, "backend", "memory")
        default_ttl = getattr(config, "default_ttl", 300)
        memory_max_size = getattr(config, "memory_max_size", 1000)
        redis_config = getattr(config, "redis", None)
        if redis_config is not None:
            redis_url = getattr(redis_config, "url", None)
            redis_key_prefix = getattr(redis_config, "key_prefix", "forge:")
            redis_max_connections = getattr(redis_config, "max_connections", 10)

    if backend_type == "redis":
        from forge.cache.backends.redis import RedisBackend

        url = redis_url or "redis://localhost:6379/0"
        redis_backend = RedisBackend(
            redis_url=url,
            key_prefix=redis_key_prefix,
            max_connections=redis_max_connections,
            default_ttl=default_ttl,
        )
        await redis_backend.connect()
        self._backend = redis_backend
    else:
        from forge.cache.backends.memory import MemoryBackend

        self._backend = MemoryBackend(
            max_size=memory_max_size,
            default_ttl=default_ttl,
        )
teardown async 
teardown() -> None

Teardown the cache module.

Source code in src/forge/cache/module.py 
82
83
84
85
86
87
88
async def teardown(self) -> None:
    """Teardown the cache module."""
    set_cache_module(None)
    if self._backend:
        await self._backend.close()
        self._backend = None
    self._runtime = None

Functions:

cached 

cached(ttl: int | None = None, key: str | None = None, namespace: str | None = None, backend: CacheBackend | None = None) -> Callable[[F], F]

Decorator for caching async function results.

Parameters:

Name Type Description Default
ttl int | None

Time-to-live in seconds. Defaults to cache module default_ttl.

 None
key str | None

Cache key template. Supports {arg_name} interpolation from function args.

 None
namespace str | None

Optional namespace prefix for the cache key.

 None
backend CacheBackend | None

Optional custom backend to use instead of the active module.

 None
Source code in src/forge/cache/decorators.py 
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
def cached(
    ttl: int | None = None,
    key: str | None = None,
    namespace: str | None = None,
    backend: CacheBackend | None = None,
) -> Callable[[F], F]:
    """
    Decorator for caching async function results.

    Args:
        ttl: Time-to-live in seconds. Defaults to cache module default_ttl.
        key: Cache key template. Supports {arg_name} interpolation from function args.
        namespace: Optional namespace prefix for the cache key.
        backend: Optional custom backend to use instead of the active module.
    """

    def decorator(func: F) -> F:
        if not inspect.iscoroutinefunction(func):
            raise TypeError(
                f"@cached requires an async function. {func.__qualname__} is synchronous."
            )

        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            # 1. Resolve cache client/backend
            b = backend
            cm = None
            if b is None:
                cm = get_cache_module()

            # 2. If no cache available, call the function transparently
            if b is None and cm is None:
                return await func(*args, **kwargs)

            # 3. Resolve key
            cache_key = _resolve_key(func, key, namespace, args, kwargs)

            # 4. Check cache
            if cm is not None:
                cached_value = await cm.get(cache_key)
            else:
                assert b is not None  # noqa: S101
                cached_value = await b.get(cache_key)

            if cached_value is not None:
                return cached_value

            # 5. Call function and cache result
            result = await func(*args, **kwargs)

            if cm is not None:
                await cm.set(cache_key, result, ttl=ttl)
            else:
                assert b is not None  # noqa: S101
                await b.set(cache_key, result, ttl=ttl)

            return result

        # Attach invalidation helper to the wrapped function
        async def invalidate(*args: Any, **kwargs: Any) -> bool:
            b = backend
            cm = None
            if b is None:
                cm = get_cache_module()

            if b is None and cm is None:
                return False

            cache_key = _resolve_key(func, key, namespace, args, kwargs)

            if cm is not None:
                return await cm.delete(cache_key)
            assert b is not None  # noqa: S101
            return await b.delete(cache_key)

        setattr(wrapper, "invalidate", invalidate)  # noqa: B010
        return cast("F", wrapper)

    return decorator

invalidate async 

invalidate(key: str) -> bool

Convenience helper to invalidate a cache key in the active cache module.

Source code in src/forge/cache/__init__.py 
 8
 9
10
11
12
13
14
15
async def invalidate(key: str) -> bool:
    """Convenience helper to invalidate a cache key in the active cache module."""
    from forge.cache._state import get_cache_module

    cm = get_cache_module()
    if cm is not None:
        return await cm.delete(key)
    return False