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