feat(flags): Add feature flags enum

Author: tcjennings

src/lsst/cmservice/common/flags.py

@@ -0,0 +1,60 @@
+import os
+from enum import IntFlag, auto
+from functools import reduce
+from typing import Any
+
+from pydantic import Field, ValidationInfo, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Features(IntFlag):
+    """A flag enum for setting specific application behaviors through feature
+    flags.
+    """
+
+    API_V1 = auto()
+    API_V2 = auto()
+    DAEMON_CAMPAIGNS = auto()
+    DAEMON_NODES = auto()
+    DAEMON_V1 = auto()
+    DAEMON_V2 = auto()
+    WEBAPP_V1 = auto()
+
+
+class EnabledFeatures(BaseSettings):
+    """Pydantic Settings class for managing the enabled features of an
+    application."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="FEATURE_",
+        case_sensitive=False,
+        extra="allow",
+    )
+
+    enabled: Features = Field(
+        description="A Flag Enum for enabled application features.", default=Features(0)
+    )
+
+    @field_validator("enabled")
+    @classmethod
+    def determine_enabled_features(cls, data: Any, info: ValidationInfo) -> Any:
+        """Check all environment variables according to the `env_prefix` of the
+        model config and set/unset feature flags for the matching feature.
+        """
+        if (env_prefix := cls.model_config.get("env_prefix")) is None:
+            return data
+        enabled = set()
+        disabled = set()
+        for feature_env_var in filter(lambda k: k.startswith(env_prefix), os.environ.keys()):
+            try:
+                if os.getenv(feature_env_var, "false").strip().lower() in ("1", "t", "true", "yes", "on"):
+                    enabled.add(Features[feature_env_var.replace(env_prefix, "")])
+                else:
+                    disabled.add(Features[feature_env_var.replace(env_prefix, "")])
+            except KeyError:
+                # no matching feature
+                pass
+
+        data = reduce(lambda x, y: x | y, enabled, data)
+        data = reduce(lambda x, y: x & ~y, disabled, data)
+        return data

tests/common/test_flags.py

@@ -0,0 +1,25 @@
+import pytest
+
+from lsst.cmservice.common.flags import EnabledFeatures, Features
+
+
+def test_feature_flags(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Empty features have nothing set
+    features = EnabledFeatures()
+    assert Features.DAEMON_V1 not in features.enabled
+    assert features.enabled.value == 0
+
+    # Setting features by env var should result in their being set in the
+    # flags enum. Nonexistent features should not be an error.
+    monkeypatch.setenv("FEATURE_DAEMON_V2", "1")
+    monkeypatch.setenv("FEATURE_DAEMON_CAMPAIGNS", "on")
+    monkeypatch.setenv("FEATURE_NO_SUCH_FEATURE", "true")
+    features = EnabledFeatures()
+    assert Features.DAEMON_V2 in features.enabled
+    assert Features.DAEMON_CAMPAIGNS in features.enabled
+
+    # Disabling a feature by env var should result in their not being set in
+    # the flags enum, overriding the default or initial value
+    monkeypatch.setenv("FEATURE_WEBAPP_V1", "0")
+    features = EnabledFeatures(enabled=Features.WEBAPP_V1)
+    assert Features.WEBAPP_V1 not in features.enabled