Coverage for src/lexigram/features/backends/env.py: 27%

1"""Environment-variable-based flag provider. 

2 

3Reads feature flags from environment variables matching the pattern 

4``{prefix}{FLAG_NAME}`` where the value encodes the flag type and its 

5parameters. The parsed result is cached in-process; call 

6:meth:`EnvProvider.invalidate` to force a re-read. 

7 

8Supported value formats: 

9 

10* ``true`` / ``false`` — :attr:`~lexigram.feature_flags.types.FlagType.BOOLEAN` 

11* ``percentage:N`` — :attr:`~lexigram.feature_flags.types.FlagType.PERCENTAGE` rollout 

12* ``users:user1,user2,...`` — :attr:`~lexigram.feature_flags.types.FlagType.USER_LIST` 

13""" 

14 

15from __future__ import annotations 

16 

17import os 

18 

19from lexigram.features.backends.base import AbstractFlagProvider 

20from lexigram.features.constants import ENV_PREFIX 

21from lexigram.features.types import Flag, FlagContext, FlagEvaluation, FlagType 

22 

23 

24class EnvProvider(AbstractFlagProvider): 

25 """Feature flag provider that reads from environment variables. 

26 

27 Flag names are lowercased by stripping the prefix. 

28 """ 

29 

30 def __init__(self, prefix: str = ENV_PREFIX) -> None: 

31 self.prefix = prefix 

32 self._cache: dict[str, Flag] | None = None 

33 

34 # -- AbstractFlagProvider interface ------------------------------------- 

35 

36 async def get_flag_definition(self, name: str) -> Flag | None: 

37 """Return the parsed :class:`Flag` for *name*, or *None*.""" 

38 flags = await self.get_all_flags() 

39 return flags.get(name) 

40 

41 async def get_all_flags(self) -> dict[str, Flag]: 

42 """Return all flags parsed from the current environment.""" 

43 if self._cache is not None: 

44 return self._cache 

45 flags: dict[str, Flag] = {} 

46 for key, value in os.environ.items(): 

47 if key.startswith(self.prefix): 

48 flag_name = key[len(self.prefix) :].lower() 

49 flag = self._parse_flag(flag_name, value.strip()) 

50 if flag is not None: 

51 flags[flag_name] = flag 

52 self._cache = flags 

53 return flags 

54 

55 # -- Synchronous bridge ------------------------------------------------- 

56 

57 def evaluate_sync( 

58 self, 

59 name: str, 

60 context: FlagContext | None = None, 

61 ) -> FlagEvaluation: 

62 """Synchronous evaluation using the last-loaded env snapshot.""" 

63 flags = self._cache or {} 

64 flag = flags.get(name) 

65 if flag is None: 

66 return FlagEvaluation( 

67 flag_name=name, 

68 enabled=False, 

69 reason="flag_not_found", 

70 value=False, 

71 ) 

72 return self._evaluate_flag(flag, context or FlagContext()) 

73 

74 def invalidate(self) -> None: 

75 """Discard the cached snapshot so the next read re-reads env vars.""" 

76 self._cache = None 

77 

78 # -- Parsing ------------------------------------------------------------ 

79 

80 def _parse_flag(self, name: str, raw: str) -> Flag | None: 

81 """Parse a raw environment variable value into a :class:`Flag`. 

82 

83 Returns *None* for values that cannot be interpreted. 

84 """ 

85 lower = raw.lower() 

86 

87 if lower in ("true", "1", "yes", "on"): 

88 return Flag(name=name, type=FlagType.BOOLEAN, enabled=True) 

89 

90 if lower in ("false", "0", "no", "off"): 

91 return Flag(name=name, type=FlagType.BOOLEAN, enabled=False) 

92 

93 if lower.startswith("percentage:"): 

94 try: 

95 pct = int(lower.split(":", 1)[1]) 

96 return Flag(name=name, type=FlagType.PERCENTAGE, percentage=pct) 

97 except ValueError: 

98 return None 

99 

100 if lower.startswith("users:"): 

101 user_list = [ 

102 u.strip() for u in raw.split(":", 1)[1].split(",") if u.strip() 

103 ] 

104 return Flag(name=name, type=FlagType.USER_LIST, user_list=user_list) 

105 

106 return None 

107 

108 

109__all__ = ["EnvProvider"]