1import logging
  2from abc import ABC, abstractmethod
  3from functools import cached_property
  4from typing import Any
  5
  6from opentelemetry import trace
  7from opentelemetry.semconv._incubating.attributes.feature_flag_attributes import (
  8    FEATURE_FLAG_KEY,
  9    FEATURE_FLAG_PROVIDER_NAME,
 10    FEATURE_FLAG_RESULT_REASON,
 11    FEATURE_FLAG_RESULT_VALUE,
 12    FeatureFlagResultReasonValues,
 13)
 14
 15from plain.runtime import settings
 16from plain.utils import timezone
 17
 18from . import exceptions
 19from .utils import coerce_key
 20
 21logger = logging.getLogger(__name__)
 22tracer = trace.get_tracer("plain.flags")
 23
 24
 25class Flag(ABC):
 26    @abstractmethod
 27    def get_key(self) -> Any:
 28        """
 29        Determine a unique key for this instance of the flag.
 30        This should be a quick operation, as it will be called on every use of the flag.
 31
 32        For convenience, you can return an instance of a Plain Model
 33        and it will be converted to a string automatically.
 34
 35        Return a falsy value if you don't want to store the flag result.
 36        """
 37        ...
 38
 39    @abstractmethod
 40    def get_value(self) -> Any:
 41        """
 42        Compute the resulting value of the flag.
 43
 44        The value needs to be JSON serializable.
 45
 46        If get_key() returns a value, this will only be called once per key
 47        and then subsequent calls will return the saved value from the DB.
 48        """
 49        ...
 50
 51    def get_db_name(self) -> str:
 52        """
 53        Should basically always be the name of the class.
 54        But this is overridable in case of renaming/refactoring/importing.
 55        """
 56        return self.__class__.__name__
 57
 58    def retrieve_or_compute_value(self) -> Any:
 59        """
 60        Retrieve the value from the DB if it exists,
 61        otherwise compute the value and save it to the DB.
 62        """
 63        from .models import Flag, FlagResult  # So Plain app is ready...
 64
 65        flag_name = self.get_db_name()
 66
 67        with tracer.start_as_current_span(
 68            f"flag {flag_name}",
 69            attributes={
 70                FEATURE_FLAG_PROVIDER_NAME: "plain.flags",
 71            },
 72        ) as span:
 73            # Create an associated DB Flag that we can use to enable/disable
 74            # and tie the results to
 75            flag_obj, _ = Flag.query.update_or_create(
 76                name=flag_name,
 77                defaults={"used_at": timezone.now()},
 78            )
 79
 80            if not flag_obj.enabled:
 81                msg = f"The {flag_obj} flag has been disabled and should either not be called, or be re-enabled."
 82                span.set_attribute(
 83                    FEATURE_FLAG_RESULT_REASON,
 84                    FeatureFlagResultReasonValues.DISABLED.value,
 85                )
 86
 87                if settings.DEBUG:
 88                    raise exceptions.FlagDisabled(msg)
 89                else:
 90                    logger.exception(msg)
 91                    # Might not be the type of return value expected! Better than totally crashing now though.
 92                    return None
 93
 94            key = self.get_key()
 95            if not key:
 96                # No key, so we always recompute the value and return it
 97                value = self.get_value()
 98
 99                span.set_attribute(
100                    FEATURE_FLAG_RESULT_REASON,
101                    FeatureFlagResultReasonValues.TARGETING_MATCH.value,
102                )
103                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(value))
104
105                return value
106
107            key = coerce_key(key)
108
109            span.set_attribute(FEATURE_FLAG_KEY, key)
110
111            try:
112                flag_result = FlagResult.query.get(flag=flag_obj, key=key)
113
114                span.set_attribute(
115                    FEATURE_FLAG_RESULT_REASON,
116                    FeatureFlagResultReasonValues.CACHED.value,
117                )
118                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(flag_result.value))
119
120                return flag_result.value
121            except FlagResult.DoesNotExist:
122                value = self.get_value()
123                flag_result = FlagResult.query.create(
124                    flag=flag_obj, key=key, value=value
125                )
126
127                span.set_attribute(
128                    FEATURE_FLAG_RESULT_REASON,
129                    FeatureFlagResultReasonValues.STATIC.value,
130                )
131                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(value))
132
133                return flag_result.value
134
135    @cached_property
136    def value(self) -> Any:
137        """
138        Cached version of retrieve_or_compute_value()
139        """
140        return self.retrieve_or_compute_value()
141
142    def __bool__(self) -> bool:
143        """
144        Allow for use in boolean expressions.
145        """
146        return bool(self.value)
147
148    def __contains__(self, item: Any) -> bool:
149        """
150        Allow for use in `in` expressions.
151        """
152        return item in self.value
153
154    def __eq__(self, other: object) -> bool:
155        """
156        Allow for use in `==` expressions.
157        """
158        return self.value == other