Coverage for src/lexigram/features/decorators/sync.py: 97%

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

2 

3Provides :func:`feature_flag_sync` and :func:`require_flag_sync` for 

4guarding synchronous functions behind a named feature flag. Only suitable 

5when the backing provider has an in-memory evaluation path. 

6""" 

7 

8from __future__ import annotations 

9 

10from collections.abc import Callable 

11import functools 

12from typing import TYPE_CHECKING, Any, TypeVar 

13 

14from lexigram.features.decorators._utils import _resolve_manager 

15from lexigram.features.exceptions import FeatureFlagDisabledError 

16 

17if TYPE_CHECKING: 

18 from lexigram.features.manager.flag_manager import FlagManager 

19 from lexigram.features.types import FlagContext 

20 

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

22 

23 

24def feature_flag_sync( 

25 name: str, 

26 *, 

27 manager: FlagManager | None = None, 

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

29 context: FlagContext | None = None, 

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

31 """Decorate a **synchronous** function so it only runs when *name* is enabled. 

32 

33 Evaluation uses :meth:`~lexigram.features.backends.base.AbstractFlagProvider.evaluate_sync` 

34 on the underlying provider. 

35 

36 Args: 

37 name: Feature flag name to check. 

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

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

40 context: Optional evaluation context. 

41 

42 Returns: 

43 A decorator for synchronous functions. 

44 """ 

45 

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

47 @functools.wraps(fn) 

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

49 mgr = _resolve_manager(manager) 

50 override = mgr.get_override_state(name) 

51 if override is True: 

52 return fn(*args, **kwargs) 

53 if override is False: 

54 if fallback is not None: 

55 return fallback(*args, **kwargs) 

56 raise FeatureFlagDisabledError(name) 

57 provider = mgr.provider 

58 result = provider.evaluate_sync(name, context) 

59 if result.enabled: 

60 return fn(*args, **kwargs) 

61 if fallback is not None: 

62 return fallback(*args, **kwargs) 

63 raise FeatureFlagDisabledError(name) 

64 

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

66 

67 return decorator 

68 

69 

70def require_flag_sync( 

71 name: str, 

72 *, 

73 manager: FlagManager | None = None, 

74 context: FlagContext | None = None, 

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

76 """Synchronous variant of :func:`~lexigram.features.decorators.require_flag`. 

77 

78 Raises :class:`~lexigram.features.exceptions.FeatureFlagDisabledError` 

79 when *name* is disabled. 

80 

81 Args: 

82 name: Feature flag name to check. 

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

84 context: Optional evaluation context. 

85 

86 Returns: 

87 A decorator for synchronous functions. 

88 """ 

89 return feature_flag_sync(name, manager=manager, context=context) 

90 

91 

92__all__ = ["feature_flag_sync", "require_flag_sync"]