1  """
 2  FeatureChecker: two-tier feature flag evaluation.
 3  
 4  Evaluation priority (highest to lowest):
 5    1. Environment variable  SAM_FEATURE_<UPPER_KEY>=true|false
 6    2. Registry default      FeatureDefinition.default
 7  """
 8  
 9  from __future__ import annotations
10  
11  import logging
12  import os
13  from typing import TYPE_CHECKING, Optional
14  
15  if TYPE_CHECKING:
16      from .registry import FeatureRegistry
17  
18  logger = logging.getLogger(__name__)
19  
20  # Prefix for environment variable overrides (e.g., SAM_FEATURE_MY_FLAG=true)
21  _FLAG_PREFIX = "SAM_FEATURE_"
22  
23  
24  class FeatureChecker:
25      """
26      Evaluates feature flags according to the two-tier priority chain.
27  
28      Parameters
29      ----------
30      registry:
31          The FeatureRegistry holding all known FeatureDefinition objects.
32      """
33  
34      def __init__(self, registry: "FeatureRegistry") -> None:
35          self._registry = registry
36  
37      def is_known_flag(self, key: str) -> bool:
38          """Return True if key is registered in the registry."""
39          return self._registry.get(key) is not None
40  
41      def is_enabled(self, key: str) -> bool:
42          """
43          Return True if the feature identified by key is currently enabled.
44  
45          Unknown keys always return False with a warning log.
46          """
47          definition = self._registry.get(key)
48          if definition is None:
49              logger.warning("is_enabled() called for unknown feature key '%s'", key)
50              return False
51  
52          env_value = self._check_env_var(key)
53          if env_value is not None:
54              logger.debug("Feature '%s' resolved from env var: %s", key, env_value)
55              return env_value
56  
57          logger.debug(
58              "Feature '%s' resolved from registry default: %s",
59              key,
60              definition.default,
61          )
62          return definition.default
63  
64      def all_flags(self) -> dict[str, bool]:
65          """Return {key: is_enabled(key)} for every registered feature."""
66          return {defn.key: self.is_enabled(defn.key) for defn in self._registry.all()}
67  
68      @property
69      def registry(self) -> "FeatureRegistry":
70          """Return the underlying FeatureRegistry."""
71          return self._registry
72  
73      def has_env_override(self, key: str) -> bool:
74          """Return True if a SAM_FEATURE_<KEY> env var is set for key."""
75          return self._check_env_var(key) is not None
76  
77      def load_from_yaml(self, yaml_path) -> None:
78          """Load additional flag definitions from a YAML file into the registry."""
79          self._registry.load_from_yaml(yaml_path)
80  
81      @staticmethod
82      def _check_env_var(key: str) -> Optional[bool]:
83          """
84          Look up SAM_FEATURE_<UPPER_KEY> in the environment.
85  
86          Returns None when the variable is absent so callers can fall through.
87          Recognised truthy values: 1, true (case-insensitive).
88          Everything else is treated as false.
89          """
90          env_key = _FLAG_PREFIX + key.upper().replace("-", "_")
91          raw = os.environ.get(env_key)
92          if raw is None:
93              return None
94          return raw.strip().lower() in {"1", "true"}