Coverage for src/lexigram/features/backends/base.py: 95%
110 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:04 +0800
1"""Abstract base class for rich flag providers.
3Defines :class:`AbstractFlagProvider` — an ABC that extends the simple
4:class:`~lexigram.contracts.feature_flags.FlagProvider` protocol with a
5richer ``evaluate`` coroutine and a complete suite of per-type evaluation
6strategy helpers.
8Concrete providers (:class:`~lexigram.feature_flags.backends.local.LocalProvider`,
9:class:`~lexigram.feature_flags.backends.env.EnvProvider`, etc.) inherit from
10this class and override ``get_flag_definition`` and ``get_all_flags``; the
11evaluation logic is shared here.
12"""
14from __future__ import annotations
16from abc import ABC, abstractmethod
17from datetime import UTC, datetime
18import hashlib
19from typing import Any
21from lexigram.features.constants import DEFAULT_ENABLED
22from lexigram.features.types import (
23 Flag,
24 FlagContext,
25 FlagEvaluation,
26 FlagType,
27)
28from lexigram.logging import get_logger
30logger = get_logger(__name__)
32_warned_empty_user_attribute_rules: set[str] = set()
35class AbstractFlagProvider(ABC):
36 """Rich abstract flag provider with full per-type evaluation logic.
38 Subclasses must implement :meth:`get_flag_definition` and
39 :meth:`get_all_flags`. All evaluation helpers and the :meth:`evaluate`
40 coroutine are provided here.
42 This class is *not* a :class:`~lexigram.contracts.feature_flags.FlagProvider`
43 protocol implementation — ``get_flag_definition`` returns a
44 :class:`~lexigram.feature_flags.types.Flag` model, not a bare boolean.
45 Use :class:`~lexigram.features.backends.local.LocalProvider`
46 when you only need the simple boolean protocol.
47 """
49 # ------------------------------------------------------------------
50 # Abstract interface — subclasses must implement these two methods.
51 # ------------------------------------------------------------------
53 @abstractmethod
54 async def get_flag_definition(self, name: str) -> Flag | None:
55 """Return the full :class:`Flag` definition or *None* if not found."""
57 @abstractmethod
58 async def get_all_flags(self) -> dict[str, Flag]:
59 """Return all known flag definitions keyed by flag name."""
61 # ------------------------------------------------------------------
62 # Primary public API
63 # ------------------------------------------------------------------
65 async def evaluate(
66 self,
67 name: str,
68 context: FlagContext | None = None,
69 ) -> FlagEvaluation:
70 """Evaluate flag *name* for *context* and return a :class:`FlagEvaluation`.
72 Returns a ``flag_not_found`` evaluation (``enabled=False``) when the
73 flag does not exist, so callers can safely treat unknown flags the same
74 as disabled ones.
75 """
76 flag = await self.get_flag_definition(name)
77 if flag is None:
78 return FlagEvaluation(
79 flag_name=name,
80 enabled=False,
81 reason="flag_not_found",
82 value=False,
83 )
84 return self._evaluate_flag(flag, context or FlagContext())
86 def evaluate_sync(
87 self,
88 name: str,
89 context: FlagContext | None = None,
90 ) -> FlagEvaluation:
91 """Synchronous evaluation — only valid when the store is in-memory.
93 Subclasses that back their data with a sync structure should override
94 this if they need to avoid the event loop. The default raises
95 ``NotImplementedError`` to prevent accidental blocking I/O.
96 """
97 raise NotImplementedError(
98 f"{type(self).__name__} does not support synchronous evaluation",
99 )
101 # ------------------------------------------------------------------
102 # FlagProvider protocol bridge
103 # ------------------------------------------------------------------
105 async def get_flag(
106 self,
107 name: str,
108 *,
109 default: bool = False,
110 context: dict[str, Any] | None = None,
111 ) -> bool:
112 """Asynchronous boolean access (primary) — delegates to evaluate."""
113 ctx = _dict_to_context(context)
114 result = await self.evaluate(name, ctx)
115 if result.reason == "flag_not_found":
116 return default
117 return result.enabled
119 def get_flag_sync(
120 self,
121 name: str,
122 *,
123 default: bool = False,
124 context: dict[str, Any] | None = None,
125 ) -> bool:
126 """Synchronous boolean access — delegates to evaluate_sync."""
127 ctx = _dict_to_context(context)
128 try:
129 result = self.evaluate_sync(name, ctx)
130 except NotImplementedError:
131 return default
133 if result.reason == "flag_not_found":
134 return default
135 return result.enabled
137 async def get_variant(
138 self,
139 name: str,
140 *,
141 default: str = "",
142 context: dict[str, Any] | None = None,
143 ) -> str:
144 """Asynchronous variant access (primary) — delegates to evaluate."""
145 ctx = _dict_to_context(context)
146 result = await self.evaluate(name, ctx)
147 if result.reason == "flag_not_found":
148 return default
149 if isinstance(result.value, str):
150 return result.value
151 return default
153 def get_variant_sync(
154 self,
155 name: str,
156 *,
157 default: str = "",
158 context: dict[str, Any] | None = None,
159 ) -> str:
160 """Synchronous variant access — delegates to evaluate_sync."""
161 ctx = _dict_to_context(context)
162 try:
163 result = self.evaluate_sync(name, ctx)
164 except NotImplementedError:
165 return default
166 if isinstance(result.value, str):
167 return result.value
168 return default
170 # ------------------------------------------------------------------
171 # Per-type evaluation helpers — shared by all concrete providers.
172 # ------------------------------------------------------------------
174 def _evaluate_flag(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
175 """Dispatch evaluation based on flag type."""
176 if not flag.enabled:
177 return FlagEvaluation(
178 flag_name=flag.name,
179 enabled=False,
180 reason="flag_disabled",
181 value=False,
182 )
184 dispatch = {
185 FlagType.BOOLEAN: self._evaluate_boolean,
186 FlagType.PERCENTAGE: self._evaluate_percentage,
187 FlagType.USER_LIST: self._evaluate_user_list,
188 FlagType.USER_ATTRIBUTE: self._evaluate_user_attribute,
189 FlagType.TIME_BASED: self._evaluate_time_based,
190 FlagType.VARIANT: self._evaluate_variant,
191 }
192 handler = dispatch.get(flag.type)
193 if handler is None:
194 return FlagEvaluation(
195 flag_name=flag.name,
196 enabled=False,
197 reason=f"unsupported_type:{flag.type}",
198 value=False,
199 )
200 return handler(flag, context)
202 def _evaluate_boolean(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
203 """Evaluate a simple boolean flag."""
204 return FlagEvaluation(
205 flag_name=flag.name,
206 enabled=flag.enabled,
207 reason="boolean",
208 value=flag.enabled,
209 )
211 def _evaluate_percentage(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
212 """Evaluate a percentage-rollout flag.
214 Hash is computed from ``{user_id}:{flag_name}`` so each flag produces
215 an independent, deterministic assignment per user.
216 """
217 bucket_input = f"{context.user_id or ''}:{flag.name}"
218 bucket = (
219 int(
220 hashlib.md5(bucket_input.encode(), usedforsecurity=False).hexdigest(),
221 16,
222 )
223 % 100
224 )
225 enabled = bucket < flag.percentage
226 return FlagEvaluation(
227 flag_name=flag.name,
228 enabled=enabled,
229 reason="percentage_rollout",
230 value=enabled,
231 )
233 def _evaluate_user_list(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
234 """Evaluate a flag restricted to an explicit user-ID list."""
235 enabled = context.user_id is not None and context.user_id in flag.user_list
236 return FlagEvaluation(
237 flag_name=flag.name,
238 enabled=enabled,
239 reason="user_list",
240 value=enabled,
241 )
243 def _evaluate_user_attribute(
244 self,
245 flag: Flag,
246 context: FlagContext,
247 ) -> FlagEvaluation:
248 """Evaluate a flag gate by matching all required user attribute pairs.
250 An empty rule set evaluates disabled (fail-closed) because an
251 unconfigured rule is a misconfiguration, not a grant-all.
252 """
253 if not flag.user_attributes:
254 if flag.name not in _warned_empty_user_attribute_rules:
255 _warned_empty_user_attribute_rules.add(flag.name)
256 logger.warning(
257 "user_attribute_empty_rule_denied",
258 flag=flag.name,
259 )
260 return FlagEvaluation(
261 flag_name=flag.name,
262 enabled=DEFAULT_ENABLED,
263 reason="user_attribute_empty_rule_denied",
264 value=DEFAULT_ENABLED,
265 )
266 attrs = context.user_attributes or {}
267 enabled = all(attrs.get(k) == v for k, v in flag.user_attributes.items())
268 return FlagEvaluation(
269 flag_name=flag.name,
270 enabled=enabled,
271 reason="user_attribute",
272 value=enabled,
273 )
275 def _evaluate_time_based(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
276 """Evaluate a time-windowed flag against UTC now (or context timestamp)."""
277 if context.timestamp is not None:
278 now = datetime.fromtimestamp(context.timestamp, tz=UTC)
279 else:
280 now = datetime.now(UTC)
282 after_start = flag.start_time is None or now >= flag.start_time
283 before_end = flag.end_time is None or now <= flag.end_time
284 enabled = after_start and before_end
286 return FlagEvaluation(
287 flag_name=flag.name,
288 enabled=enabled,
289 reason="time_based",
290 value=enabled,
291 )
293 def _evaluate_variant(self, flag: Flag, context: FlagContext) -> FlagEvaluation:
294 """Assign a variant deterministically using weighted bucket hashing.
296 Variants are sorted by name for deterministic ordering, then a
297 cumulative weight range is used to pick the variant for this user.
298 """
299 if not flag.variants:
300 variant = flag.default_variant or ""
301 return FlagEvaluation(
302 flag_name=flag.name,
303 enabled=bool(variant),
304 reason="variant_no_config",
305 value=variant or False,
306 )
308 bucket_input = f"{context.user_id or ''}:{flag.name}"
309 bucket = (
310 int(
311 hashlib.md5(bucket_input.encode(), usedforsecurity=False).hexdigest(),
312 16,
313 )
314 % 100
315 )
317 cumulative = 0
318 chosen: str | None = None
319 for variant_name in sorted(flag.variants):
320 cumulative += flag.variants[variant_name]
321 if bucket < cumulative:
322 chosen = variant_name
323 break
325 if chosen is None:
326 chosen = flag.default_variant or next(iter(sorted(flag.variants)), "")
328 return FlagEvaluation(
329 flag_name=flag.name,
330 enabled=True,
331 reason="variant",
332 value=chosen,
333 )
336# ---------------------------------------------------------------------------
337# Helpers
338# ---------------------------------------------------------------------------
341def _dict_to_context(d: dict[str, Any] | None) -> FlagContext | None:
342 """Convert a plain dict context to a :class:`FlagContext` if given."""
343 if d is None:
344 return None
345 return FlagContext(
346 user_id=d.get("user_id"),
347 user_attributes=d.get("user_attributes"),
348 session_id=d.get("session_id"),
349 request_id=d.get("request_id"),
350 timestamp=d.get("timestamp"),
351 custom={
352 k: v
353 for k, v in d.items()
354 if k
355 not in {
356 "user_id",
357 "user_attributes",
358 "session_id",
359 "request_id",
360 "timestamp",
361 }
362 }
363 or None,
364 )
367__all__ = ["AbstractFlagProvider"]