Coverage for src/lexigram/features/decorators/feature_flag.py: 100%

1"""Async feature-flag–gated decorators. 

2 

3Provides :func:`feature_flag` and :func:`require_flag` for guarding async 

4functions behind a named feature flag. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections.abc import Callable 

10import functools 

11from typing import TYPE_CHECKING, Any, TypeVar 

12 

13from lexigram.features.decorators._utils import _is_coroutinefunction, _resolve_manager 

14from lexigram.features.exceptions import FeatureFlagDisabledError 

15 

16if TYPE_CHECKING: 

17 from lexigram.features.manager.flag_manager import FlagManager 

18 from lexigram.features.types import FlagContext 

19 

20F = TypeVar("F", bound=Callable[..., Any]) 

21 

22 

23def feature_flag( 

24 name: str, 

25 *, 

26 manager: FlagManager | None = None, 

27 fallback: Callable[..., Any] | None = None, 

28 context: FlagContext | None = None, 

29) -> Callable[[F], F]: 

30 """Decorate an async function so it only runs when *name* is enabled. 

31 

32 When the flag is *disabled*: 

33 

34 * If *fallback* is provided it is called with the same arguments and its 

35 return value is returned. 

36 * Otherwise :class:`~lexigram.features.exceptions.FeatureFlagDisabledError` 

37 is raised. 

38 

39 Args: 

40 name: Feature flag name to check. 

41 manager: Explicit :class:`~lexigram.features.manager.FlagManager`; resolved 

42 via DI or a default when ``None``. 

43 fallback: Optional callable invoked when the flag is disabled. 

44 context: Optional evaluation context passed to the manager. 

45 

46 Returns: 

47 A decorator for async functions. 

48 """ 

49 

50 def decorator(fn: F) -> F: 

51 @functools.wraps(fn) 

52 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

53 mgr = _resolve_manager(manager) 

54 enabled = await mgr.is_enabled(name, context) 

55 if enabled: 

56 return await fn(*args, **kwargs) 

57 if fallback is not None: 

58 if _is_coroutinefunction(fallback): 

59 return await fallback(*args, **kwargs) 

60 return fallback(*args, **kwargs) 

61 raise FeatureFlagDisabledError(name) 

62 

63 return wrapper # type: ignore[return-value] 

64 

65 return decorator 

66 

67 

68def require_flag( 

69 name: str, 

70 *, 

71 manager: FlagManager | None = None, 

72 context: FlagContext | None = None, 

73) -> Callable[[F], F]: 

74 """Decorate an async function to raise when *name* is disabled. 

75 

76 Semantically identical to :func:`feature_flag` with no fallback, but 

77 the intent is clearer when used as an access guard. 

78 

79 Args: 

80 name: Feature flag name to check. 

81 manager: Explicit manager; resolved via DI or a default when ``None``. 

82 context: Optional evaluation context. 

83 

84 Returns: 

85 A decorator for async functions. 

86 """ 

87 return feature_flag(name, manager=manager, context=context) 

88 

89 

90__all__ = ["feature_flag", "require_flag"]