Coverage for src/lexigram/features/backends/cache.py: 43%
49 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 01:53 +0800
1"""Cache-backed feature flag provider.
3Stores flag definitions in a :class:`~lexigram.contracts.cache.protocols.CacheBackendProtocol`
4(e.g. Redis, Memcached, or in-memory). Each flag is serialised as JSON under a
5namespaced key so that multiple applications can share the same cache without
6key collisions.
8Limitations
9-----------
10* :meth:`CacheBackendFlagProvider.list_flags` returns an empty list because
11 most cache backends (including Redis ``CacheBackendProtocol`` implementations) do not
12 expose a key-scan API through the standard protocol. Applications that need
13 to list all flags should maintain a separate index or use a provider that
14 supports enumeration (e.g. :class:`~lexigram.features.backends.local.LocalProvider`).
15"""
17from __future__ import annotations
19from datetime import UTC
20from typing import TYPE_CHECKING, Any
22from lexigram import serialization as json
23from lexigram.features.exceptions import FlagNotFoundError
24from lexigram.features.types import Flag, FlagType
25from lexigram.logging import get_logger
27if TYPE_CHECKING:
28 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
30logger = get_logger(__name__)
33class CacheBackendFlagProvider:
34 """Feature-flag provider backed by a :class:`CacheBackendProtocol`.
36 Flags are stored as JSON-encoded objects in the cache under keys of the
37 form ``{prefix}{name}``. Each entry is given an independent TTL so that
38 flags expire and are treated as absent after the TTL elapses.
40 Args:
41 cache: The cache backend to use for flag storage and retrieval.
42 ttl: Time-to-live in seconds for each cached flag entry. Defaults
43 to 3 600 seconds (1 hour).
44 prefix: Key prefix applied to every flag name to avoid collisions
45 with other cache users. Defaults to ``"lexigram:features:flag:"``.
46 """
48 def __init__(
49 self,
50 cache: CacheBackendProtocol,
51 ttl: int = 3600,
52 prefix: str = "lexigram:features:flag:",
53 ) -> None:
54 self._cache = cache
55 self._ttl = ttl
56 self._prefix = prefix
58 # ------------------------------------------------------------------
59 # Internal helpers
60 # ------------------------------------------------------------------
62 def _key(self, name: str) -> str:
63 """Build the namespaced cache key for *name*."""
64 return f"{self._prefix}{name}"
66 @staticmethod
67 def _flag_to_dict(flag: Flag) -> dict[str, Any]:
68 """Serialise a :class:`Flag` to a JSON-compatible dict."""
69 return {
70 "name": flag.name,
71 "type": flag.type.value,
72 "enabled": flag.enabled,
73 "description": flag.description,
74 "percentage": flag.percentage,
75 "user_list": flag.user_list,
76 "user_attributes": flag.user_attributes,
77 "start_time": flag.start_time.isoformat() if flag.start_time else None,
78 "end_time": flag.end_time.isoformat() if flag.end_time else None,
79 "variants": flag.variants,
80 "default_variant": flag.default_variant,
81 "metadata": flag.metadata,
82 }
84 @staticmethod
85 def _dict_to_flag(data: dict[str, Any]) -> Flag:
86 """Deserialise a dict produced by :meth:`_flag_to_dict` into a :class:`Flag`."""
87 from datetime import datetime
89 def _parse_dt(value: str | None) -> datetime | None:
90 if value is None:
91 return None
92 return datetime.fromisoformat(value).replace(tzinfo=UTC)
94 return Flag(
95 name=data["name"],
96 type=FlagType(data.get("type", FlagType.BOOLEAN)),
97 enabled=data.get("enabled", True),
98 description=data.get("description", ""),
99 percentage=data.get("percentage", 0),
100 user_list=data.get("user_list", []),
101 user_attributes=data.get("user_attributes", {}),
102 start_time=_parse_dt(data.get("start_time")),
103 end_time=_parse_dt(data.get("end_time")),
104 variants=data.get("variants", {}),
105 default_variant=data.get("default_variant", ""),
106 metadata=data.get("metadata", {}),
107 )
109 # ------------------------------------------------------------------
110 # Public API
111 # ------------------------------------------------------------------
113 async def get_flag(self, name: str) -> Flag:
114 """Return the :class:`Flag` definition stored under *name*.
116 Args:
117 name: The flag name to look up.
119 Returns:
120 The :class:`Flag` stored in the cache.
122 Raises:
123 FlagNotFoundError: If no flag with *name* exists in the cache.
124 """
125 raw_result = await self._cache.get(self._key(name))
126 if not raw_result.is_ok():
127 raise FlagNotFoundError(name)
128 raw = raw_result.unwrap()
129 if raw is None:
130 raise FlagNotFoundError(name)
131 data: dict[str, Any] = json.loads(raw) if isinstance(raw, (str, bytes)) else raw
132 return self._dict_to_flag(data)
134 async def add_flag(self, flag: Flag) -> None:
135 """Write *flag* to the cache with the configured TTL.
137 Args:
138 flag: The :class:`Flag` definition to persist.
139 """
140 payload = json.dumps(self._flag_to_dict(flag))
141 await self._cache.set(self._key(flag.name), payload, ttl=self._ttl)
142 logger.debug("cache_flag_added", flag=flag.name, ttl=self._ttl)
144 async def remove_flag(self, name: str) -> None:
145 """Delete the flag with *name* from the cache.
147 A no-op if the flag does not exist (the cache backend returns False
148 from ``delete``, which is silently ignored here).
150 Args:
151 name: The flag name to remove.
152 """
153 await self._cache.delete(self._key(name))
154 logger.debug("cache_flag_removed", flag=name)
156 async def list_flags(self) -> list[Flag]:
157 """Return all known flags.
159 .. note::
160 This method always returns an **empty list** because the standard
161 :class:`~lexigram.contracts.cache.protocols.CacheBackendProtocol` protocol
162 does not expose a key-scan or key-enumeration operation (e.g. Redis
163 ``SCAN``). If you need to enumerate flags, maintain a separate
164 index in the cache (a set under a dedicated key) or use a provider
165 that natively supports enumeration such as
166 :class:`~lexigram.features.backends.local.LocalProvider`.
168 Returns:
169 Always an empty :class:`list`.
170 """
171 logger.debug(
172 "cache_flag_list_unsupported",
173 message=(
174 "CacheBackendFlagProvider.list_flags() returns [] — "
175 "key enumeration is not part of the CacheBackendProtocol protocol."
176 ),
177 )
178 return []
180 async def evaluate(self, name: str, context: dict[str, Any] | None = None) -> bool:
181 """Evaluate whether the flag *name* is enabled.
183 Fetches the flag from the cache and returns its ``enabled`` field.
184 This is a simple boolean check; for richer evaluation (percentage
185 rollout, user-list, variant, etc.) use a full
186 :class:`~lexigram.features.backends.base.AbstractFlagProvider`.
188 Args:
189 name: The flag to evaluate.
190 context: Optional context dict (not used by this provider; included
191 for API consistency with other providers).
193 Returns:
194 ``True`` if the flag exists and is enabled; ``False`` if the flag
195 is disabled.
197 Raises:
198 FlagNotFoundError: If the flag does not exist in the cache.
199 """
200 flag = await self.get_flag(name)
201 return flag.enabled
204__all__ = ["CacheBackendFlagProvider"]