Coverage for src/lexigram/features/backends/local.py: 100%

1"""In-memory, code-defined flag provider. 

2 

3Suitable for flags defined at application startup (e.g. loaded from a 

4config file), single-node applications, and as a base layer in a 

5:class:`~lexigram.feature_flags.backends.chained.ChainedProvider`. 

6""" 

7 

8from __future__ import annotations 

9 

10from datetime import UTC, datetime 

11 

12from lexigram.features.backends.base import AbstractFlagProvider 

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

14 

15 

16class LocalProvider(AbstractFlagProvider): 

17 """In-memory provider backed by a ``dict[str, Flag]``. 

18 

19 Dynamic update methods (:meth:`add_flag`, :meth:`remove_flag`, 

20 :meth:`set_enabled`) mutate the store safely within a single process. 

21 """ 

22 

23 def __init__(self, flags: dict[str, Flag] | None = None) -> None: 

24 self._flags: dict[str, Flag] = dict(flags or {}) 

25 

26 # -- AbstractFlagProvider interface ------------------------------------- 

27 

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

29 """Return the :class:`Flag` definition for *name*, or *None*.""" 

30 return self._flags.get(name) 

31 

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

33 """Return a snapshot copy of all defined flags.""" 

34 return dict(self._flags) 

35 

36 # -- Synchronous evaluation bridge -------------------------------------- 

37 

38 def evaluate_sync( 

39 self, 

40 name: str, 

41 context: FlagContext | None = None, 

42 ) -> FlagEvaluation: 

43 """Synchronous evaluation — safe because the backing store is in-memory.""" 

44 flag = self._flags.get(name) 

45 if flag is None: 

46 return FlagEvaluation( 

47 flag_name=name, 

48 enabled=False, 

49 reason="flag_not_found", 

50 value=False, 

51 ) 

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

53 

54 # -- Mutation helpers --------------------------------------------------- 

55 

56 def add_flag(self, flag: Flag) -> None: 

57 """Add or replace a flag definition.""" 

58 flag.updated_at = datetime.now(UTC) 

59 self._flags[flag.name] = flag 

60 

61 def remove_flag(self, name: str) -> bool: 

62 """Remove a flag by name; returns *True* if it existed.""" 

63 return self._flags.pop(name, None) is not None 

64 

65 def set_enabled(self, name: str, enabled: bool) -> bool: 

66 """Toggle the master enabled switch; returns *True* if the flag exists.""" 

67 if name not in self._flags: 

68 return False 

69 self._flags[name].enabled = enabled 

70 self._flags[name].updated_at = datetime.now(UTC) 

71 return True 

72 

73 # -- Convenience helpers (also satisfy MutableFlagProvider protocol) ---- 

74 

75 async def set_flag(self, name: str, value: bool) -> None: 

76 """Store a boolean flag (async primary).""" 

77 self.set_flag_sync(name, value) 

78 

79 def set_flag_sync(self, name: str, value: bool) -> None: 

80 """Store a boolean flag (sync).""" 

81 self.add_flag(Flag(name=name, type=FlagType.BOOLEAN, enabled=value)) 

82 

83 async def set_variant(self, name: str, variant: str) -> None: 

84 """Store a variant flag (async primary).""" 

85 self.set_variant_sync(name, variant) 

86 

87 def set_variant_sync(self, name: str, variant: str) -> None: 

88 """Store a variant flag (sync).""" 

89 self.add_flag( 

90 Flag( 

91 name=name, 

92 type=FlagType.VARIANT, 

93 enabled=True, 

94 default_variant=variant, 

95 ), 

96 ) 

97 

98 def clear(self) -> None: 

99 """Remove all stored flags and variants.""" 

100 self._flags.clear() 

101 

102 

103__all__ = ["LocalProvider"]