Coverage for src/lexigram/admin/services/feature_flags.py: 0%
93 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Feature flags service for lexigram-admin.
3Provides :class:`AdminFeatureFlagService` — a thin admin-specific wrapper
4around :class:`~lexigram.contracts.feature_flags.protocols.FlagManagerProtocol`
5that enforces the ``admin.`` namespace prefix and exposes admin-specific
6decorators and helpers.
8FWK-03: AdminFeatureFlagService implementation.
9"""
11from __future__ import annotations
13from dataclasses import dataclass
14from functools import wraps
15from typing import Any
17from lexigram.contracts.exceptions import ConfigurationError, InfrastructureError
18from lexigram.contracts.feature_flags.protocols import FlagManagerProtocol
19from lexigram.di.decorators import inject
21# ============================================================================
22# Admin Feature Flag Configuration
23# ============================================================================
26@dataclass
27class AdminFeatureConfig:
28 """Configuration for admin features.
30 Each boolean field represents a feature that can be toggled.
31 """
33 # Core features
34 soft_delete: bool = True
35 audit_logging: bool = True
36 bulk_operations: bool = True
37 export_data: bool = True
38 import_data: bool = True
40 # UI features
41 dark_mode: bool = True
42 sidebar_collapse: bool = True
43 table_column_resize: bool = True
44 inline_editing: bool = False
46 # Advanced features
47 custom_dashboards: bool = False
48 api_explorer: bool = False
49 webhook_management: bool = False
50 scheduled_tasks: bool = False
52 # Beta features
53 ai_assistant: bool = False
54 predictive_search: bool = False
57# ============================================================================
58# Admin Feature Flag Service
59# ============================================================================
62@inject
63class AdminFeatureFlagService:
64 """Feature flag service for the admin panel.
66 Delegates flag evaluation to a FlagManager while enforcing the ``admin.``
67 prefix namespace and seeding initial values from :class:`AdminFeatureConfig`.
69 The manager implementation is supplied through DI via contracts.
71 Example::
73 service = AdminFeatureFlagService(manager, AdminFeatureConfig())
74 if await service.is_enabled("soft_delete"):
75 ...
76 """
78 def __init__(
79 self,
80 manager: FlagManagerProtocol,
81 config: AdminFeatureConfig | None = None,
82 ) -> None:
83 """Initialize the service.
85 Args:
86 manager: Feature flag manager whose provider is a LocalProvider.
87 Requires ``lexigram-features`` to be installed.
88 config: Optional configuration controlling initial flag values.
89 Defaults to :class:`AdminFeatureConfig`.
91 Raises:
92 InfrastructureError: If no feature-flag manager binding is available.
93 """
94 if manager is None:
95 raise InfrastructureError(
96 "AdminFeatureFlagService requires a FlagManagerProtocol binding.",
97 )
98 self._manager = manager
99 self._config = config or AdminFeatureConfig()
100 self._register_config_flags(self._config)
102 # ------------------------------------------------------------------
103 # Internal helpers
104 # ------------------------------------------------------------------
106 def _normalize(self, flag_name: str) -> str:
107 """Ensure *flag_name* carries the ``admin.`` prefix."""
108 if not flag_name.startswith("admin."):
109 return f"admin.{flag_name}"
110 return flag_name
112 def _register_config_flags(self, config: AdminFeatureConfig) -> None:
113 """Validate feature configuration payload shape."""
114 for field_name in config.__dataclass_fields__:
115 if not isinstance(getattr(config, field_name), bool):
116 msg = f"AdminFeatureConfig field '{field_name}' must be bool."
117 raise ConfigurationError(msg)
119 # ------------------------------------------------------------------
120 # Public API
121 # ------------------------------------------------------------------
123 def is_enabled_sync(
124 self,
125 flag_name: str,
126 context: dict[str, Any] | None = None,
127 default: bool = False,
128 ) -> bool:
129 """Synchronously check whether a feature flag is enabled.
131 Safe to call outside an async context because the backing store is
132 always in-memory. For the async primary, use :meth:`is_enabled`.
134 Args:
135 flag_name: Flag name (with or without the ``admin.`` prefix).
136 context: Optional evaluation context.
137 default: Fallback value when the flag is not found.
139 Returns:
140 True if the flag is enabled.
141 """
142 flag_name = self._normalize(flag_name)
143 short_name = flag_name.removeprefix("admin.")
144 if short_name in self._config.__dataclass_fields__:
145 return bool(getattr(self._config, short_name))
146 return default
148 async def is_enabled(
149 self,
150 flag_name: str,
151 context: dict[str, Any] | None = None,
152 default: bool = False,
153 ) -> bool:
154 """Check if a feature flag is enabled (async primary).
156 Args:
157 flag_name: Flag name (with or without the ``admin.`` prefix).
158 context: Optional evaluation context.
159 default: Fallback value when the flag is not found.
161 Returns:
162 True if the flag is enabled.
163 """
164 flag_name = self._normalize(flag_name)
165 value = await self._manager.get_value(flag_name, default, context=context)
166 return bool(value)
168 def set_flag(self, flag_name: str, enabled: bool) -> None:
169 """Set a feature flag value.
171 Args:
172 flag_name: Flag name (with or without the ``admin.`` prefix).
173 enabled: Whether the flag should be enabled.
174 """
175 raise ConfigurationError(
176 "AdminFeatureFlagService.set_flag() is unsupported in contract-only mode. "
177 "Set flags via your configured FlagManagerProtocol provider.",
178 )
180 async def get_all_flags(self) -> dict[str, bool]:
181 """Return all admin feature flags.
183 Returns:
184 Mapping of short names (without ``admin.`` prefix) to enabled state.
185 """
186 all_flags = await self._manager.get_all_flags()
187 result: dict[str, bool] = {}
188 for key, evaluation in all_flags.items():
189 if key.startswith("admin."):
190 result[key.removeprefix("admin.")] = bool(evaluation.value)
191 return result
193 def require_flag(self, flag_name: str) -> None:
194 """Raise if a flag is not enabled.
196 Args:
197 flag_name: Flag to check.
199 Raises:
200 FeatureDisabledError: If the flag is not enabled.
201 """
202 if not self.is_enabled_sync(flag_name):
203 raise FeatureDisabledError(flag_name)
205 def get_variant(
206 self,
207 flag_name: str,
208 context: dict[str, Any] | None = None,
209 default: str = "",
210 ) -> str:
211 """Return the variant string for a VARIANT-type flag.
213 Args:
214 flag_name: Flag name (with or without the ``admin.`` prefix).
215 context: Optional evaluation context.
216 default: Fallback value when no variant is stored.
218 Returns:
219 The variant string or *default*.
220 """
221 _ = self._normalize(flag_name)
222 _ = context
223 return default
226# ============================================================================
227# Module-level helpers
228# ============================================================================
231async def get_feature_flag_service(
232 context: Any | None = None,
233) -> AdminFeatureFlagService:
234 """Resolve the admin feature flag service from the DI container."""
235 from lexigram.admin.lib.di import get_admin_resolver
237 resolver = get_admin_resolver(context)
238 return await resolver.resolve(AdminFeatureFlagService)
241def __getattr__(name: str) -> Any:
242 if name == "feature_flag_service":
243 raise AttributeError(
244 "feature_flag_service is now async. Use: await get_feature_flag_service()",
245 )
246 raise AttributeError(f"module {__name__} has no attribute {name}")
249# ============================================================================
250# Admin-specific exception
251# ============================================================================
254class FeatureDisabledError(ConfigurationError):
255 """Raised when a required admin feature is disabled."""
257 _code: str = "LEX_ERR_ADMIN_027"
259 def __init__(self, flag_name: str) -> None:
260 self.flag_name = flag_name
261 super().__init__(
262 f"Feature '{flag_name}' is disabled",
263 details={"flag_name": flag_name},
264 )
267# ============================================================================
268# Admin-specific decorators and helpers
269# ============================================================================
272def require_feature(flag_name: str) -> Any:
273 """Decorator: raise :class:`FeatureDisabledError` if *flag_name* is disabled.
275 Example::
277 @require_feature("bulk_operations")
278 async def bulk_delete(request):
279 ...
280 """
282 def decorator(func) -> Any:
283 @wraps(func)
284 async def wrapper(*args, **kwargs) -> Any:
285 service = await get_feature_flag_service()
286 service.require_flag(flag_name)
287 return await func(*args, **kwargs)
289 return wrapper
291 return decorator
294async def feature_enabled(flag_name: str, default: bool = False) -> bool:
295 """Check if an admin feature flag is enabled.
297 Example::
299 if await feature_enabled("dark_mode"):
300 ...
301 """
302 service = await get_feature_flag_service()
303 return await service.is_enabled(flag_name, default=default)
306__all__ = [
307 "AdminFeatureConfig",
308 "AdminFeatureFlagService",
309 "FeatureDisabledError",
310 "feature_enabled",
311 "require_feature",
312]