pyapp-org/pyapp

View on GitHub
src/pyapp/feature_flags.py

Summary

Maintainability
A
35 mins
Test Coverage
"""
Feature Flags
~~~~~~~~~~~~~

Runtime configurable feature flags.

Define flags in code either in a code block:

.. code-block:: python

    if feature_flags.get("MY-FLAG"):
        pass

Or as an A/B switch:

.. code-block:: python

    instance = feature_flags.a_or_b("MY-FLAG", option_a="foo", option_b="bar")

.. tip:: ``option_a`` and ``option_b`` parameters can also be callables.

Or by using as a decorator:

.. code-block:: python

    @feature_flags.if_enabled("MY-FLAG", default=True)
    def my_feature(arg):
        pass


Define flags in the applications ``default_settings``:

.. code-block:: python

    FEATURE_FLAGS = {
        "MY-FLAG": True,
        "OTHER-FLAG": False,
    }

The flag can then be enabled/disabled in the environment::

    PYAPP_FLAG_MY_FLAG = yes
    PYAPP_FLAG_OTHER_FLAG = off

.. note:: The flag name is translated into upper-case with underscores

Or from the command line::

    my_app --enable-flag MY-FLAG --disable-flag OTHER-FLAG

"""
import logging
from functools import wraps
from os import getenv
from typing import Any, Callable, Dict, Optional, TypeVar, Union

from pyapp.conf import settings
from pyapp.utils import text_to_bool

LOGGER = logging.getLogger(__name__)
ENV_TRANSLATE = str.maketrans(
    " -abcdefghijklmnopqrstuvwxyz", "__ABCDEFGHIJKLMNOPQRSTUVWXYZ"
)
_F = TypeVar("_F", bound=Callable[..., Any])


class FeatureFlags:
    """
    Feature flags object that caches resolved flags.

    Flags are resolved in the following order:

    - CLI
    - Environment Variables
    - Settings

    """

    __slots__ = ("_cache",)

    def __init__(self):
        self._cache: Dict[str, bool] = {}

    @staticmethod
    def _resolve_from_environment(flag: str) -> Optional[bool]:
        """
        Attempt to resolve from environment.
        """
        key = f"{settings.FEATURE_FLAG_PREFIX}{flag.translate(ENV_TRANSLATE)}"
        LOGGER.debug("Resolving flag %r from environment variable %s", flag, key)
        value = getenv(key, None)
        if value:
            return text_to_bool(value)
        return None

    @staticmethod
    def _resolve_from_settings(flag: str) -> Optional[bool]:
        LOGGER.debug("Resolving flag %r from settings", flag)
        return settings.FEATURE_FLAGS.get(flag, None)

    def _resolve(self, flag: str, default: bool) -> bool:
        value = self._resolve_from_environment(flag)
        if value is not None:
            LOGGER.info("Flag %r resolved from environment: %s", flag, value)
            return value

        value = self._resolve_from_settings(flag)
        if value is not None:
            LOGGER.info("Flag %r resolved from settings: %s", flag, value)
            return value

        return default

    def _get(self, flag: str, default: bool) -> bool:
        try:
            return self._cache[flag]
        except KeyError:
            value = self._cache[flag] = self._resolve(flag, default)
            return value

    def set(self, flag: str, value: bool):
        """
        Set the state of the flag
        """
        self._cache[flag] = value

    def get(self, flag: str, *, default: bool = False):
        """
        Get the state of a flag

        :param flag: Name of flag
        :param default: Default flag state

        """
        return self._get(flag, default)

    def a_or_b(
        self,
        flag: str,
        option_a: Union[Any, Callable[[], Any]],
        option_b: Union[Any, Callable[[], Any]],
        *,
        default: bool = False,
    ):
        """
        Use one of two values (or results of callables),

        :param flag: Name of flag
        :param option_a: A value or callable that provides a value; used if flag is True
        :param option_b: B value or callable that provides a value; used if flag is False
        :param default: Default flag state

        """
        if self._get(flag, default):
            return option_a() if callable(option_a) else option_a
        return option_b() if callable(option_b) else option_b

    def if_enabled(
        self, flag: str, *, default: bool = False, disabled_return: Any = None
    ) -> Callable[[_F], _F]:
        """
        Decorator that is enabled by flag.

        :param flag: Name of flag
        :param default: Default flag state
        :param disabled_return: Return value if flag is disabled

        """

        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                if self._get(flag, default):
                    return func(*args, **kwargs)
                return disabled_return

            return wrapper

        return decorator


DEFAULT = FeatureFlags()
get = DEFAULT.get  # pylint: disable=invalid-name
if_enabled = DEFAULT.if_enabled  # pylint: disable=invalid-name