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            # Resolve the key first so it's set on the span regardless of
 74            # which path we take below — including the disabled path, where
 75            # dashboards filtering by key still need to see the evaluation.
 76            key = self.get_key()
 77            if key:
 78                key = coerce_key(key)
 79                span.set_attribute(FEATURE_FLAG_KEY, key)
 80
 81            # Create an associated DB Flag that we can use to enable/disable
 82            # and tie the results to
 83            flag_obj, _ = Flag.query.update_or_create(
 84                name=flag_name,
 85                defaults={"used_at": timezone.now()},
 86            )
 87
 88            if not flag_obj.enabled:
 89                msg = f"The {flag_obj} flag has been disabled and should either not be called, or be re-enabled."
 90                span.set_attribute(
 91                    FEATURE_FLAG_RESULT_REASON,
 92                    FeatureFlagResultReasonValues.DISABLED.value,
 93                )
 94
 95                if settings.DEBUG:
 96                    raise exceptions.FlagDisabled(msg)
 97                else:
 98                    logger.exception(msg)
 99                    # Might not be the type of return value expected! Better than totally crashing now though.
100                    return None
101
102            if not key:
103                # No key, so we always recompute the value and return it
104                value = self.get_value()
105
106                span.set_attribute(
107                    FEATURE_FLAG_RESULT_REASON,
108                    FeatureFlagResultReasonValues.TARGETING_MATCH.value,
109                )
110                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(value))
111
112                return value
113
114            try:
115                flag_result = FlagResult.query.get(flag=flag_obj, key=key)
116
117                span.set_attribute(
118                    FEATURE_FLAG_RESULT_REASON,
119                    FeatureFlagResultReasonValues.CACHED.value,
120                )
121                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(flag_result.value))
122
123                return flag_result.value
124            except FlagResult.DoesNotExist:
125                value = self.get_value()
126                flag_result = FlagResult.query.create(
127                    flag=flag_obj, key=key, value=value
128                )
129
130                # Per OTel semconv, `targeting_match` is "dynamic evaluation,
131                # such as a rule or specific user-targeting" — `get_value()`
132                # ran with this key. `static` would mean "no dynamic
133                # evaluation," which doesn't apply here.
134                span.set_attribute(
135                    FEATURE_FLAG_RESULT_REASON,
136                    FeatureFlagResultReasonValues.TARGETING_MATCH.value,
137                )
138                span.set_attribute(FEATURE_FLAG_RESULT_VALUE, str(value))
139
140                return flag_result.value
141
142    @cached_property
143    def value(self) -> Any:
144        """
145        Cached version of retrieve_or_compute_value()
146        """
147        return self.retrieve_or_compute_value()
148
149    def __bool__(self) -> bool:
150        """
151        Allow for use in boolean expressions.
152        """
153        return bool(self.value)
154
155    def __contains__(self, item: Any) -> bool:
156        """
157        Allow for use in `in` expressions.
158        """
159        return item in self.value
160
161    def __eq__(self, other: object) -> bool:
162        """
163        Allow for use in `==` expressions.
164        """
165        return self.value == other