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