Coverage for src / lexigram / contracts / feature_flags / protocols.py: 100%
14 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""Feature flag protocols for the framework.
3These protocols live in ``lexigram.contracts`` so that other packages may
4rely on the abstract interface rather than importing concrete
5implementations from ``lexigram.feature_flags`` or the enterprise extension.
6"""
8from __future__ import annotations
10from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
12if TYPE_CHECKING:
13 from lexigram.contracts.feature_flags.models import FlagEvaluation, FlagValue
16@runtime_checkable
17class FlagProviderProtocol(Protocol):
18 """Protocol for reading feature flags.
20 * Supports synchronous and asynchronous access. Implementations may be
21 backed by in-memory maps, redis, remote services, etc.
22 * Context parameter is provided for user/tenant/environment lookup.
23 * Variant resolution is part of the contract, allowing A/B testing or
24 percentage-based flags.
25 * Write methods are deliberately excluded; see :class:`MutableFlagProviderProtocol`
26 for providers that support updates.
27 * Async methods are the primary entry points (no suffix). Sync variants
28 carry a ``_sync`` suffix.
29 """
31 async def get_flag(
32 self,
33 name: str,
34 *,
35 default: bool = False,
36 context: dict[str, Any] | None = None,
37 ) -> bool: # pragma: no cover - protocol
38 """Asynchronously evaluate the boolean value of a flag.
40 This is the **primary** entry point for all applications.
41 ``context`` may contain arbitrary data used by the provider to make a
42 decision (e.g. user id, tenant id, request headers).
43 """
45 def get_flag_sync(
46 self,
47 name: str,
48 *,
49 default: bool = False,
50 context: dict[str, Any] | None = None,
51 ) -> bool: # pragma: no cover - protocol
52 """Synchronously evaluate the boolean value of a flag.
54 Only suitable when the backing store is fully in-memory and no I/O is
55 required. Prefer :meth:`get_flag` in async contexts.
56 """
58 async def get_variant(
59 self,
60 name: str,
61 *,
62 default: str = "",
63 context: dict[str, Any] | None = None,
64 ) -> str: # pragma: no cover - protocol
65 """Asynchronously return a string variant for an A/B test or multivalue flag.
67 This is the **primary** entry point. Not all providers will support
68 variants; those may simply return ``default``.
69 """
71 def get_variant_sync(
72 self,
73 name: str,
74 *,
75 default: str = "",
76 context: dict[str, Any] | None = None,
77 ) -> str: # pragma: no cover - protocol
78 """Synchronously return a string variant for an A/B test or multivalue flag.
80 Only suitable when the backing store is fully in-memory. Prefer
81 :meth:`get_variant` in async contexts.
82 """
85@runtime_checkable
86class MutableFlagProviderProtocol(FlagProviderProtocol, Protocol):
87 """Optional extension of :class:`FlagProviderProtocol` that supports writes."""
89 async def set_flag(
90 self,
91 name: str,
92 value: bool,
93 ) -> None: # pragma: no cover - protocol
94 """Asynchronously create or update a boolean flag value.
96 This is the **primary** entry point. Required for providers backed by
97 remote stores (Redis, database, etc.). Implementations with an
98 in-memory backing store may perform the operation synchronously.
99 """
101 def set_flag_sync(
102 self, name: str, value: bool
103 ) -> None: # pragma: no cover - protocol
104 """Synchronously create or update a boolean flag value.
106 Only suitable when the backing store is fully in-memory. Prefer
107 :meth:`set_flag` in async contexts.
108 """
110 async def set_variant(
111 self,
112 name: str,
113 variant: str,
114 ) -> None: # pragma: no cover - protocol
115 """Asynchronously create or update the active variant for a flag.
117 This is the **primary** entry point.
118 """
120 def set_variant_sync(
121 self,
122 name: str,
123 variant: str,
124 ) -> None: # pragma: no cover - protocol
125 """Synchronously create or update the active variant for a flag.
127 Only suitable when the backing store is fully in-memory. Prefer
128 :meth:`set_variant` in async contexts.
129 """
132@runtime_checkable
133class FlagManagerProtocol(Protocol):
134 """Manages the lifecycle of feature flag providers and evaluates flags.
136 The manager chains multiple ``FlagProviderProtocol`` instances in order of priority.
137 Evaluation short-circuits to the first provider that can resolve the flag.
138 """
140 def add_provider(self, provider: FlagProviderProtocol, priority: int = 50) -> None:
141 """Register a flag provider at the given priority level.
143 Higher priority values are queried first. Providers with equal
144 priority are queried in registration order.
146 Args:
147 provider: The flag provider to register.
148 priority: Resolution priority (higher = queried first).
149 """
150 ...
152 async def is_enabled(
153 self,
154 key: str,
155 context: dict[str, Any] | None = None,
156 ) -> bool:
157 """Evaluate a boolean feature flag.
159 Args:
160 key: The flag identifier.
161 context: Optional evaluation context (user, tenant, etc.).
163 Returns:
164 True if the flag is enabled, False otherwise.
166 Raises:
167 FlagNotFoundError: If no provider can resolve the flag.
168 """
169 ...
171 async def get_value(
172 self,
173 key: str,
174 default: FlagValue,
175 context: dict[str, Any] | None = None,
176 ) -> FlagValue:
177 """Evaluate a feature flag and return its resolved value.
179 Args:
180 key: The flag identifier.
181 default: Fallback value when the flag cannot be resolved.
182 context: Optional evaluation context.
184 Returns:
185 The resolved FlagValue, or ``default`` if not found.
186 """
187 ...
189 async def evaluate(
190 self,
191 key: str,
192 context: dict[str, Any] | None = None,
193 ) -> FlagEvaluation:
194 """Return the full evaluation result including metadata.
196 Args:
197 key: The flag identifier.
198 context: Optional evaluation context.
200 Returns:
201 A FlagEvaluation with value, type, reason, and metadata.
203 Raises:
204 FlagNotFoundError: If no provider can resolve the flag.
205 """
206 ...
208 async def get_all_flags(
209 self,
210 context: dict[str, Any] | None = None,
211 ) -> dict[str, FlagEvaluation]:
212 """Return evaluations for all known flags.
214 Args:
215 context: Optional evaluation context applied to every flag.
217 Returns:
218 Mapping of flag key to its FlagEvaluation.
219 """
220 ...
223__all__ = [
224 "FlagManagerProtocol",
225 "FlagProviderProtocol",
226 "MutableFlagProviderProtocol",
227]