Coverage for src/lexigram/features/manager/flag_manager.py: 87%
145 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:04 +0800
1"""Central feature flag manager implementation.
3:class:`FlagManager` is the primary entry point for evaluating feature flags
4in application code. It wraps any :class:`~lexigram.features.backends.base.AbstractFlagProvider`
5and adds in-process TTL caching, runtime overrides, change listeners, variant
6support, and a configurable default-enabled fallback.
7"""
9from __future__ import annotations
11import asyncio
12from dataclasses import dataclass, field
13from datetime import UTC, datetime
14import time
15from typing import TYPE_CHECKING, Any
17from lexigram.concurrency.task_utils import create_tracked_task
18from lexigram.features.backends.local import LocalProvider
19from lexigram.features.types import FlagContext, FlagEvaluation
20from lexigram.logging import get_logger
22logger = get_logger(__name__)
24if TYPE_CHECKING:
25 from lexigram.contracts.events.protocols import EventBusProtocol
26 from lexigram.features.backends.base import AbstractFlagProvider
27 from lexigram.features.manager.types import (
28 AsyncFlagChangeListener,
29 FlagChangeListener,
30 )
33@dataclass
34class FlagAuditEntry:
35 """Records a single flag override change for compliance and observability.
37 Attributes:
38 flag_name: The name of the feature flag that was changed.
39 actor: Optional identifier of the entity that made the change.
40 old_value: The override state before the change (``None`` if no prior override).
41 new_value: The override state after the change.
42 timestamp: UTC datetime when the change was recorded.
43 """
45 flag_name: str
46 actor: str | None
47 old_value: bool | None
48 new_value: bool
49 timestamp: datetime = field(default_factory=lambda: datetime.now(tz=UTC))
52class FlagManager:
53 """Central feature flag manager with caching, overrides, and variant support.
55 Wraps an :class:`~lexigram.features.backends.base.AbstractFlagProvider` and
56 handles cache invalidation, runtime overrides, and change notification.
58 Example::
60 provider = LocalProvider({
61 "new_checkout": Flag("new_checkout", enabled=True),
62 })
63 manager = FlagManager(provider, cache_ttl=60)
65 ctx = FlagContext(user_id="user-123")
66 if await manager.is_enabled("new_checkout", ctx):
67 ...
68 """
70 def __init__(
71 self,
72 provider: AbstractFlagProvider | None = None,
73 *,
74 cache_ttl: int = 300,
75 default_enabled: bool = False,
76 event_bus: EventBusProtocol | None = None,
77 ) -> None:
78 self._provider: AbstractFlagProvider = provider or LocalProvider()
79 self._cache_ttl = cache_ttl
80 self._default_enabled = default_enabled
81 self._event_bus: EventBusProtocol | None = event_bus
82 self._listeners: list[FlagChangeListener] = []
83 self._async_listeners: list[AsyncFlagChangeListener] = []
84 self._disabled_flags: set[str] = set()
85 self._enabled_flags: set[str] = set()
86 self._cache: dict[str, tuple[FlagEvaluation, float]] = {}
87 self._background_tasks: set[asyncio.Task[Any]] = set()
88 self._audit_log: list[FlagAuditEntry] = []
90 # ------------------------------------------------------------------
91 # Public read API
92 # ------------------------------------------------------------------
94 async def is_enabled(
95 self,
96 name: str,
97 context: FlagContext | None = None,
98 *,
99 default: bool | None = None,
100 ) -> bool:
101 """Return whether *name* is enabled for *context*.
103 Runtime overrides take precedence over the provider result.
105 Args:
106 name: Flag name.
107 context: Optional evaluation context.
108 default: Override the instance's ``default_enabled`` for this call.
110 Returns:
111 True if the flag is enabled.
112 """
113 if name in self._enabled_flags:
114 return True
115 if name in self._disabled_flags:
116 return False
118 evaluation = await self.evaluate(name, context)
119 if evaluation.reason == "flag_not_found":
120 if default is not None:
121 return default
122 return self._default_enabled
123 return evaluation.enabled
125 async def evaluate(
126 self,
127 name: str,
128 context: FlagContext | None = None,
129 ) -> FlagEvaluation:
130 """Evaluate *name* for *context*, using the TTL cache when available.
132 Args:
133 name: Flag name.
134 context: Optional evaluation context.
136 Returns:
137 A :class:`~lexigram.features.types.FlagEvaluation` result.
138 """
139 cache_key = self._get_cache_key(name, context)
140 if self._cache_ttl > 0 and cache_key in self._cache:
141 evaluation, ts = self._cache[cache_key]
142 if time.monotonic() - ts < self._cache_ttl:
143 return evaluation
145 try:
146 evaluation = await self._provider.evaluate(name, context)
147 except Exception as exc: # noqa: BLE001
148 logger.error(
149 "flag_provider_error",
150 flag=name,
151 error=str(exc),
152 )
153 return FlagEvaluation(
154 flag_name=name,
155 enabled=self._default_enabled,
156 reason="provider_error",
157 )
159 if self._cache_ttl > 0:
160 self._cache[cache_key] = (evaluation, time.monotonic())
161 return evaluation
163 async def get_variant(
164 self,
165 name: str,
166 context: FlagContext | None = None,
167 *,
168 default: str = "",
169 ) -> str:
170 """Return the variant string for a VARIANT-type flag.
172 Returns *default* if the flag is not found, disabled, or is not a
173 VARIANT flag.
175 Args:
176 name: Flag name.
177 context: Optional evaluation context.
178 default: Fallback value when no variant string is available.
180 Returns:
181 The variant string or *default*.
182 """
183 evaluation = await self.evaluate(name, context)
184 if isinstance(evaluation.value, str):
185 return evaluation.value
186 return default
188 def add_provider(self, provider: AbstractFlagProvider, priority: int = 50) -> None:
189 """Register a flag provider at the given priority level."""
190 # For simplicity in this adaptation, just replace the default provider.
191 # A robust implementation would support multiple chained providers.
192 self._provider = provider
194 async def get_value(
195 self,
196 key: str,
197 default: object,
198 context: FlagContext | None = None,
199 ) -> object:
200 """Evaluate a feature flag and return its resolved value."""
201 from lexigram.features.backends.base import _dict_to_context
203 ctx = _dict_to_context(context) if isinstance(context, dict) else context
204 evaluation = await self.evaluate(key, ctx)
205 if evaluation.reason == "flag_not_found":
206 return default
207 return (
208 evaluation.value
209 if hasattr(evaluation, "value") and evaluation.value is not None
210 else default
211 )
213 async def get_all_flags(
214 self,
215 context: FlagContext | None = None,
216 ) -> dict[str, FlagEvaluation]:
217 """Return evaluations for all known flags.
219 Definitions come from the provider; each is evaluated against
220 *context* here so the mapping honours the
221 ``dict[str, FlagEvaluation]`` contract instead of leaking raw
222 ``Flag`` definitions.
223 """
224 if not hasattr(self._provider, "get_all_flags"):
225 return {}
228 definitions: dict[str, Any] = await self._provider.get_all_flags()
229 evaluations: dict[str, FlagEvaluation] = {}
230 for name in definitions:
231 evaluation = await self._provider.evaluate(name, context)
232 evaluations[name] = evaluation
233 return evaluations
235 # ------------------------------------------------------------------
236 # Runtime overrides
237 # ------------------------------------------------------------------
239 def get_override_state(self, name: str) -> bool | None:
240 """Return the current runtime override for *name*.
242 Returns:
243 ``True`` if force-enabled, ``False`` if force-disabled,
244 ``None`` if no override is active.
245 """
246 if name in self._enabled_flags:
247 return True
248 if name in self._disabled_flags:
249 return False
250 return None
252 def enable(self, name: str, *, actor: str | None = None) -> None:
253 """Force-enable *name* regardless of provider result."""
254 old_value: bool | None = self.get_override_state(name)
255 self._enabled_flags.add(name)
256 self._disabled_flags.discard(name)
257 self._audit_log.append(
258 FlagAuditEntry(
259 flag_name=name, actor=actor, old_value=old_value, new_value=True
260 )
261 )
262 self._notify_listeners(name, old_value is True, True, actor=actor)
264 def disable(self, name: str, *, actor: str | None = None) -> None:
265 """Force-disable *name* regardless of provider result."""
266 old_value: bool | None = self.get_override_state(name)
267 self._disabled_flags.add(name)
268 self._enabled_flags.discard(name)
269 self._audit_log.append(
270 FlagAuditEntry(
271 flag_name=name, actor=actor, old_value=old_value, new_value=False
272 )
273 )
274 self._notify_listeners(name, old_value is True, False, actor=actor)
276 def set_override(
277 self, name: str, enabled: bool, *, actor: str | None = None
278 ) -> None:
279 """Convenience wrapper that calls :meth:`enable` or :meth:`disable`."""
280 if enabled:
281 self.enable(name, actor=actor)
282 else:
283 self.disable(name)
285 def clear_override(self, name: str) -> None:
286 """Remove any runtime override for *name*, restoring provider control."""
287 self._enabled_flags.discard(name)
288 self._disabled_flags.discard(name)
290 def get_audit_log(self) -> list[FlagAuditEntry]:
291 """Return a copy of all recorded flag change audit entries.
293 Each entry captures the flag name, actor, old value, new value, and
294 the UTC timestamp of the change. Entries are ordered oldest-first.
296 Returns:
297 List of :class:`FlagAuditEntry` instances representing every
298 :meth:`enable` / :meth:`disable` / :meth:`set_override` call
299 since this manager was created.
300 """
301 return list(self._audit_log)
303 # ------------------------------------------------------------------
304 # Cache management
305 # ------------------------------------------------------------------
307 async def clear_cache(self) -> None:
308 """Flush all cached evaluations."""
309 self._cache.clear()
311 def clear_cache_for(self, name: str) -> None:
312 """Flush cached evaluations for a single flag name."""
313 keys = [k for k in self._cache if k.startswith(f"flag:{name}")]
314 for k in keys:
315 del self._cache[k]
317 def _get_cache_key(self, name: str, context: FlagContext | None) -> str:
318 if context is None:
319 return f"flag:{name}"
320 ctx_hash = context.context_hash()
321 if not ctx_hash:
322 return f"flag:{name}"
323 return f"flag:{name}:ctx:{ctx_hash}"
325 # ------------------------------------------------------------------
326 # Change listeners
327 # ------------------------------------------------------------------
329 def add_listener_sync(self, fn: FlagChangeListener) -> None:
330 """Register a sync callback invoked when a flag override changes."""
331 self._listeners.append(fn)
333 def remove_listener_sync(self, fn: FlagChangeListener) -> None:
334 """Deregister a previously registered sync listener."""
335 self._listeners.remove(fn)
337 def add_listener(self, fn: AsyncFlagChangeListener) -> None:
338 """Register an async callback invoked when a flag override changes."""
339 self._async_listeners.append(fn)
341 def remove_listener(self, fn: AsyncFlagChangeListener) -> None:
342 """Deregister a previously registered async listener."""
343 self._async_listeners.remove(fn)
345 def _notify_listeners(
346 self,
347 name: str,
348 old_enabled: bool,
349 new_enabled: bool,
350 actor: str | None = None,
351 ) -> None:
352 """Call all sync listeners synchronously; schedule async listeners."""
353 for fn in self._listeners:
354 fn(name, old_enabled, new_enabled)
355 try:
356 asyncio.get_running_loop()
357 if self._async_listeners:
358 for afn in self._async_listeners:
359 create_tracked_task(
360 afn(name, old_enabled, new_enabled),
361 self._background_tasks,
362 name=f"flag_listener_{name}",
363 )
364 if self._event_bus is not None:
365 from lexigram.features.events import FlagChangeEvent
367 event = FlagChangeEvent(
368 flag_name=name,
369 old_enabled=old_enabled,
370 new_enabled=new_enabled,
371 actor=actor,
372 )
373 create_tracked_task(
374 self._event_bus.publish(event),
375 self._background_tasks,
376 name=f"flag_event_{name}",
377 )
378 except RuntimeError:
379 pass # No running event loop; async listeners silently skipped.
381 # ------------------------------------------------------------------
382 # Introspection
383 # ------------------------------------------------------------------
385 @property
386 def provider(self) -> AbstractFlagProvider:
387 """The underlying flag provider instance."""
388 return self._provider
391__all__ = ["FlagManager"]