Cookbook: Pattern: Feature Toggles with AOP¶

# feature_toggle_lib/decorator.py
from enum import Enum
from typing import Callable

# Metadata key used to store toggle info on functions
FEATURE_TOGGLE_META = "_pico_feature_toggle_meta"

class Mode(str, Enum):
    """Behavior when the feature is disabled."""
    EXCEPTION = "exception"      # Raise RuntimeError
    RETURN_NONE = "return_none"  # Return None silently

def feature_toggle(*, name: str, mode: Mode = Mode.RETURN_NONE) -> Callable[[Callable], Callable]:
    """Decorator to mark a method as controlled by a feature toggle.
    It attaches metadata to the function without wrapping it.
    """
    def decorator(func: Callable) -> Callable:
        setattr(func, FEATURE_TOGGLE_META, {"name": name, "mode": mode})
        return func
    return decorator

# feature_toggle_lib/registry.py
import os
from pico_ioc import component

@component
class FeatureToggleRegistry:
    """Checks if features are enabled, typically via environment variables."""
    def __init__(self):
        # Read a comma-separated list of DISABLED features
        disabled_str = os.environ.get("PICO_FEATURES_DISABLED", "")
        self._disabled_set = {
            feature.strip().lower()
            for feature in disabled_str.split(",")
            if feature.strip()
        }
        print(f"[Registry] Disabled features: {self._disabled_set}")

    def is_enabled(self, feature_name: str) -> bool:
        """Check if a feature toggle is currently enabled."""
        return feature_name.lower() not in self._disabled_set

# feature_toggle_lib/interceptor.py
from typing import Any, Callable
from pico_ioc import component, MethodInterceptor, MethodCtx
from .registry import FeatureToggleRegistry
from .decorator import FEATURE_TOGGLE_META, Mode

@component
class FeatureToggleInterceptor(MethodInterceptor):
    def __init__(self, registry: FeatureToggleRegistry):
        # Inject the registry
        self.registry = registry
        print("[Interceptor] FeatureToggleInterceptor initialized.")

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        # Try to read metadata directly from the current method
        toggle_meta = getattr(ctx.method, FEATURE_TOGGLE_META, None)

        # Fallback: read from the attribute on the declaring class, if present
        if not toggle_meta:
            try:
                original_func = getattr(ctx.cls, ctx.name)
                toggle_meta = getattr(original_func, FEATURE_TOGGLE_META, None)
            except AttributeError:
                toggle_meta = None

        if not toggle_meta:
            # This method isn't feature-toggled; proceed normally
            return call_next(ctx)

        feature_name = toggle_meta["name"]

        if self.registry.is_enabled(feature_name):
            # Feature is ON, proceed with the original call
            print(f"[Interceptor] Feature '{feature_name}' is ENABLED. Proceeding.")
            return call_next(ctx)
        else:
            # Feature is OFF, apply the disabled behavior
            print(f"[Interceptor] Feature '{feature_name}' is DISABLED. Blocking.")
            mode = toggle_meta["mode"]
            if mode == Mode.EXCEPTION:
                raise RuntimeError(f"Feature '{feature_name}' is currently disabled.")
            else:  # Mode.RETURN_NONE
                return None

# feature_toggle_lib/__init__.py
from .decorator import feature_toggle, Mode
from .registry import FeatureToggleRegistry
from .interceptor import FeatureToggleInterceptor

__all__ = [
    "feature_toggle", "Mode",
    "FeatureToggleRegistry", "FeatureToggleInterceptor",
]

# my_app/services.py
from pico_ioc import component, intercepted_by
from feature_toggle_lib import (
    feature_toggle, Mode, FeatureToggleInterceptor
)

@component
class MyService:

    # Apply the toggle decorator AND the interceptor
    @feature_toggle(name="new-reporting", mode=Mode.EXCEPTION)
    @intercepted_by(FeatureToggleInterceptor)
    def generate_report(self, user_id: int) -> dict:
        print("[MyService] Generating complex new report...")
        # ... actual reporting logic ...
        return {"report_id": 123, "user": user_id}

    # Apply toggle with different mode
    @feature_toggle(name="experimental-feature", mode=Mode.RETURN_NONE)
    @intercepted_by(FeatureToggleInterceptor)
    def try_experimental(self) -> str | None:
        print("[MyService] Trying experimental feature...")
        return "Experimental Success!"

    # This method is not toggled and not intercepted
    def always_available(self) -> str:
        print("[MyService] Running essential function.")
        return "Always OK"

# Note: You could apply @intercepted_by at the class level
# if all methods should potentially be checked by the interceptor.
# @component
# @intercepted_by(FeatureToggleInterceptor)
# class MyService:
#     @feature_toggle(...)
#     def method1(...): ...
#     def method2(...): ... # Interceptor runs, finds no metadata, proceeds

# main.py
import os
from pico_ioc import init
from my_app.services import MyService

def run_app():
    print("--- Initializing Container ---")
    # Scan both the app and the feature toggle library
    container = init(modules=["my_app", "feature_toggle_lib"])
    print("--- Container Initialized ---\n")

    service = container.get(MyService)

    # --- Test Cases ---
    print("--- Testing 'always_available' ---")
    print(f"Result: {service.always_available()}\n")

    print("--- Testing 'generate_report' (new-reporting feature) ---")
    try:
        report = service.generate_report(user_id=42)
        print(f"Result: {report}\n")
    except Exception as e:
        print(f"Caught Exception: {e}\n")

    print("--- Testing 'try_experimental' (experimental-feature) ---")
    exp_result = service.try_experimental()
    print(f"Result: {exp_result}\n")


if __name__ == "__main__":
    print("="*30)
    print("RUN 1: ALL FEATURES ENABLED")
    print("="*30)
    # Ensure no features are disabled
    os.environ.pop("PICO_FEATURES_DISABLED", None)
    run_app()

    print("\n" + "="*30)
    print("RUN 2: 'new-reporting' FEATURE DISABLED")
    print("="*30)
    # Disable one feature via environment variable
    os.environ["PICO_FEATURES_DISABLED"] = "new-reporting"
    run_app()

    print("\n" + "="*30)
    print("RUN 3: BOTH FEATURES DISABLED")
    print("="*30)
    # Disable multiple features
    os.environ["PICO_FEATURES_DISABLED"] = "new-reporting,experimental-feature"
    run_app()