6782905699

Committed 07 Nov 2023 10:05AM UTC coverage: 96.905% (+0.2%) from 96.684%

Build # 6782905699

Build Type

Pull #285

github

Committed by

web-flow

Commit Message

Merge f5793db94 into 2ddb217b0

Pull Request Pull Request #285: feat: add is_feature_enabled to variant response

Run Details

5 of 6 new or added lines in 3 files covered. (83.33%)

7 existing lines in 2 files now uncovered.

908 of 937 relevant lines covered (96.91%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.07

/UnleashClient/__init__.py

# pylint: disable=invalid-name
import copy
import random
import string
import uuid
import warnings
from datetime import datetime, timezone
from typing import Callable, Optional

from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.job import Job
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.base import BaseScheduler
from apscheduler.triggers.interval import IntervalTrigger

from UnleashClient.api import register_client
from UnleashClient.constants import DISABLED_VARIATION, ETAG, METRIC_LAST_SENT_TIME
from UnleashClient.events import UnleashEvent, UnleashEventType
from UnleashClient.features import Feature
from UnleashClient.loader import load_features
from UnleashClient.periodic_tasks import (
    aggregate_and_send_metrics,
    fetch_and_load_features,
)
from UnleashClient.strategies import (
    ApplicationHostname,
    Default,
    FlexibleRollout,
    GradualRolloutRandom,
    GradualRolloutSessionId,
    GradualRolloutUserId,
    RemoteAddress,
    UserWithId,
)

from .cache import BaseCache, FileCache
from .deprecation_warnings import strategy_v2xx_deprecation_check
from .utils import LOGGER, InstanceAllowType, InstanceCounter

INSTANCES = InstanceCounter()


# pylint: disable=dangerous-default-value
class UnleashClient:
    """
    A client for the Unleash feature toggle system.

    :param url: URL of the unleash server, required.
    :param app_name: Name of the application using the unleash client, required.
    :param environment: Name of the environment using the unleash client, optional & defaults to "default".
    :param instance_id: Unique identifier for unleash client instance, optional & defaults to "unleash-client-python"
    :param refresh_interval: Provisioning refresh interval in seconds, optional & defaults to 15 seconds
    :param refresh_jitter: Provisioning refresh interval jitter in seconds, optional & defaults to None
    :param metrics_interval: Metrics refresh interval in seconds, optional & defaults to 60 seconds
    :param metrics_jitter: Metrics refresh interval jitter in seconds, optional & defaults to None
    :param disable_metrics: Disables sending metrics to unleash server, optional & defaults to false.
    :param disable_registration: Disables registration with unleash server, optional & defaults to false.
    :param custom_headers: Default headers to send to unleash server, optional & defaults to empty.
    :param custom_options: Default requests parameters, optional & defaults to empty.  Can be used to skip SSL verification.
    :param custom_strategies: Dictionary of custom strategy names : custom strategy objects.
    :param cache_directory: Location of the cache directory. When unset, FCache will determine the location.
    :param verbose_log_level: Numerical log level (https://docs.python.org/3/library/logging.html#logging-levels) for cases where checking a feature flag fails.
    :param cache: Custom cache implementation that extends UnleashClient.cache.BaseCache.  When unset, UnleashClient will use Fcache.
    :param scheduler: Custom APScheduler object.  Use this if you want to customize jobstore or executors.  When unset, UnleashClient will create it's own scheduler.
    :param scheduler_executor: Name of APSCheduler executor to use if using a custom scheduler.
    :param multiple_instance_mode: Determines how multiple instances being instantiated is handled by the SDK, when set to InstanceAllowType.BLOCK, the client constructor will fail when more than one instance is detected, when set to InstanceAllowType.WARN, multiple instances will be allowed but log a warning, when set to InstanceAllowType.SILENTLY_ALLOW, no warning or failure will be raised when instantiating multiple instances of the client. Defaults to InstanceAllowType.WARN
    :param event_callback: Function to call if impression events are enabled.  WARNING: Depending on your event library, this may have performance implications!
    """

    def __init__(
        self,
        url: str,
        app_name: str,
        environment: str = "default",
        instance_id: str = "unleash-client-python",
        refresh_interval: int = 15,
        refresh_jitter: Optional[int] = None,
        metrics_interval: int = 60,
        metrics_jitter: Optional[int] = None,
        disable_metrics: bool = False,
        disable_registration: bool = False,
        custom_headers: Optional[dict] = None,
        custom_options: Optional[dict] = None,
        custom_strategies: Optional[dict] = None,
        cache_directory: Optional[str] = None,
        project_name: Optional[str] = None,
        verbose_log_level: int = 30,
        cache: Optional[BaseCache] = None,
        scheduler: Optional[BaseScheduler] = None,
        scheduler_executor: Optional[str] = None,
        multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN,
        event_callback: Optional[Callable[[UnleashEvent], None]] = None,
    ) -> None:
        custom_headers = custom_headers or {}
        custom_options = custom_options or {}
        custom_strategies = custom_strategies or {}

        # Configuration
        self.unleash_url = url.rstrip("/")
        self.unleash_app_name = app_name
        self.unleash_environment = environment
        self.unleash_instance_id = instance_id
        self.unleash_refresh_interval = refresh_interval
        self.unleash_refresh_jitter = (
            int(refresh_jitter) if refresh_jitter is not None else None
        )
        self.unleash_metrics_interval = metrics_interval
        self.unleash_metrics_jitter = (
            int(metrics_jitter) if metrics_jitter is not None else None
        )
        self.unleash_disable_metrics = disable_metrics
        self.unleash_disable_registration = disable_registration
        self.unleash_custom_headers = custom_headers
        self.unleash_custom_options = custom_options
        self.unleash_static_context = {
            "appName": self.unleash_app_name,
            "environment": self.unleash_environment,
        }
        self.unleash_project_name = project_name
        self.unleash_verbose_log_level = verbose_log_level
        self.unleash_event_callback = event_callback

        self._do_instance_check(multiple_instance_mode)

        # Class objects
        self.features: dict = {}
        self.fl_job: Job = None
        self.metric_job: Job = None

        self.cache = cache or FileCache(
            self.unleash_app_name, directory=cache_directory
        )
        self.cache.mset({METRIC_LAST_SENT_TIME: datetime.now(timezone.utc), ETAG: ""})
        self.unleash_bootstrapped = self.cache.bootstrapped

        # Scheduler bootstrapping
        # - Figure out the Unleash executor name.
        if scheduler and scheduler_executor:
            self.unleash_executor_name = scheduler_executor
        elif scheduler and not scheduler_executor:
            raise ValueError(
                "If using a custom scheduler, you must specify a executor."
            )
        else:
            if not scheduler and scheduler_executor:
                LOGGER.warning(
                    "scheduler_executor should only be used with a custom scheduler."
                )

            self.unleash_executor_name = f"unleash_executor_{''.join(random.choices(string.ascii_uppercase + string.digits, k=6))}"

        # Set up the scheduler.
        if scheduler:
            self.unleash_scheduler = scheduler
        else:
            executors = {self.unleash_executor_name: ThreadPoolExecutor()}
            self.unleash_scheduler = BackgroundScheduler(executors=executors)

        # Mappings
        default_strategy_mapping = {
            "applicationHostname": ApplicationHostname,
            "default": Default,
            "gradualRolloutRandom": GradualRolloutRandom,
            "gradualRolloutSessionId": GradualRolloutSessionId,
            "gradualRolloutUserId": GradualRolloutUserId,
            "remoteAddress": RemoteAddress,
            "userWithId": UserWithId,
            "flexibleRollout": FlexibleRollout,
        }

        if custom_strategies:
            strategy_v2xx_deprecation_check(
                [x for x in custom_strategies.values()]
            )  # pylint: disable=R1721

        self.strategy_mapping = {**custom_strategies, **default_strategy_mapping}

        # Client status
        self.is_initialized = False

        # Bootstrapping
        if self.unleash_bootstrapped:
            load_features(
                cache=self.cache,
                feature_toggles=self.features,
                strategy_mapping=self.strategy_mapping,
            )

    def initialize_client(self, fetch_toggles: bool = True) -> None:
        """
        Initializes client and starts communication with central unleash server(s).

        This kicks off:

        * Client registration
        * Provisioning poll
        * Stats poll

        If `fetch_toggles` is `False`, feature toggle polling will be turned off
        and instead the client will only load features from the cache. This is
        usually used to cater the multi-process setups, e.g. Django, Celery,
        etc.

        This will raise an exception on registration if the URL is invalid. It is done automatically if called inside a context manager as in:

        .. code-block:: python

            with UnleashClient(
                url="https://foo.bar",
                app_name="myClient1",
                instance_id="myinstanceid"
                ) as client:
                pass
        """
        # Only perform initialization steps if client is not initialized.
        if not self.is_initialized:
            # pylint: disable=no-else-raise
            try:
                # Setup
                metrics_args = {
                    "url": self.unleash_url,
                    "app_name": self.unleash_app_name,
                    "instance_id": self.unleash_instance_id,
                    "custom_headers": self.unleash_custom_headers,
                    "custom_options": self.unleash_custom_options,
                    "features": self.features,
                    "cache": self.cache,
                }

                # Register app
                if not self.unleash_disable_registration:
                    register_client(
                        self.unleash_url,
                        self.unleash_app_name,
                        self.unleash_instance_id,
                        self.unleash_metrics_interval,
                        self.unleash_custom_headers,
                        self.unleash_custom_options,
                        self.strategy_mapping,
                    )

                if fetch_toggles:
                    job_args = {
                        "url": self.unleash_url,
                        "app_name": self.unleash_app_name,
                        "instance_id": self.unleash_instance_id,
                        "custom_headers": self.unleash_custom_headers,
                        "custom_options": self.unleash_custom_options,
                        "cache": self.cache,
                        "features": self.features,
                        "strategy_mapping": self.strategy_mapping,
                        "project": self.unleash_project_name,
                    }
                    job_func: Callable = fetch_and_load_features
                else:
                    job_args = {
                        "cache": self.cache,
                        "feature_toggles": self.features,
                        "strategy_mapping": self.strategy_mapping,
                    }
                    job_func = load_features

                job_func(**job_args)  # type: ignore
                # Start periodic jobs
                self.unleash_scheduler.start()
                self.fl_job = self.unleash_scheduler.add_job(
                    job_func,
                    trigger=IntervalTrigger(
                        seconds=int(self.unleash_refresh_interval),
                        jitter=self.unleash_refresh_jitter,
                    ),
                    executor=self.unleash_executor_name,
                    kwargs=job_args,
                )

                if not self.unleash_disable_metrics:
                    self.metric_job = self.unleash_scheduler.add_job(
                        aggregate_and_send_metrics,
                        trigger=IntervalTrigger(
                            seconds=int(self.unleash_metrics_interval),
                            jitter=self.unleash_metrics_jitter,
                        ),
                        executor=self.unleash_executor_name,
                        kwargs=metrics_args,
                    )
            except Exception as excep:
                # Log exceptions during initialization.  is_initialized will remain false.
                LOGGER.warning(
                    "Exception during UnleashClient initialization: %s", excep
                )
                raise excep
            else:
                # Set is_initialized to true if no exception is encountered.
                self.is_initialized = True
        else:
            warnings.warn(
                "Attempted to initialize an Unleash Client instance that has already been initialized."
            )

    def destroy(self) -> None:
        """
        Gracefully shuts down the Unleash client by stopping jobs, stopping the scheduler, and deleting the cache.

        You shouldn't need this too much!
        """
        self.fl_job.remove()
        if self.metric_job:
            self.metric_job.remove()
        self.unleash_scheduler.shutdown()
        self.cache.destroy()

    @staticmethod
    def _get_fallback_value(
        fallback_function: Callable, feature_name: str, context: dict
    ) -> bool:
        if fallback_function:
            fallback_value = fallback_function(feature_name, context)
        else:
            fallback_value = False

        return fallback_value

    # pylint: disable=broad-except
    def is_enabled(
        self,
        feature_name: str,
        context: Optional[dict] = None,
        fallback_function: Callable = None,
    ) -> bool:
        """
        Checks if a feature toggle is enabled.

        Notes:

        * If client hasn't been initialized yet or an error occurs, flat will default to false.

        :param feature_name: Name of the feature
        :param context: Dictionary with context (e.g. IPs, email) for feature toggle.
        :param fallback_function: Allows users to provide a custom function to set default value.
        :return: Feature flag result
        """
        context = context or {}

        base_context = self.unleash_static_context.copy()
        # Update context with static values and allow context to override environment
        base_context.update(context)
        context = base_context

        if self.unleash_bootstrapped or self.is_initialized:
            try:
                feature = self.features[feature_name]
                feature_check = feature.is_enabled(
                    context
                ) and self._dependencies_are_satisfied(feature_name, context)

                if feature.only_for_metrics:
                    return self._get_fallback_value(
                        fallback_function, feature_name, context
                    )

                try:
                    if self.unleash_event_callback and feature.impression_data:
                        event = UnleashEvent(
                            event_type=UnleashEventType.FEATURE_FLAG,
                            event_id=uuid.uuid4(),
                            context=context,
                            enabled=feature_check,
                            feature_name=feature_name,
                        )

                        self.unleash_event_callback(event)
                except Exception as excep:
                    LOGGER.log(
                        self.unleash_verbose_log_level,
                        "Error in event callback: %s",
                        excep,
                    )
                    return feature_check

                return feature_check
            except Exception as excep:
                LOGGER.log(
                    self.unleash_verbose_log_level,
                    "Returning default value for feature: %s",
                    feature_name,
                )
                LOGGER.log(
                    self.unleash_verbose_log_level,
                    "Error checking feature flag: %s",
                    excep,
                )
                # The feature doesn't exist, so create it to track metrics
                new_feature = Feature.metrics_only_feature(feature_name)
                self.features[feature_name] = new_feature

                # Use the feature's is_enabled method to count the call
                new_feature.is_enabled(context)

                return self._get_fallback_value(
                    fallback_function, feature_name, context
                )

        else:
            LOGGER.log(
                self.unleash_verbose_log_level,
                "Returning default value for feature: %s",
                feature_name,
            )
            LOGGER.log(
                self.unleash_verbose_log_level,
                "Attempted to get feature_flag %s, but client wasn't initialized!",
                feature_name,
            )
            return self._get_fallback_value(fallback_function, feature_name, context)

    # pylint: disable=broad-except
    def get_variant(self, feature_name: str, context: Optional[dict] = None) -> dict:
        """
        Checks if a feature toggle is enabled.  If so, return variant.

        Notes:

        * If client hasn't been initialized yet or an error occurs, flat will default to false.

        :param feature_name: Name of the feature
        :param context: Dictionary with context (e.g. IPs, email) for feature toggle.
        :return: Variant and feature flag status.
        """
        context = context or {}
        context.update(self.unleash_static_context)
        disabled_variation = copy.deepcopy(DISABLED_VARIATION)

        if self.unleash_bootstrapped or self.is_initialized:
            try:
                feature = self.features[feature_name]

                if not self._dependencies_are_satisfied(feature_name, context):
                    return disabled_variation

                variant_check = feature.get_variant(context)

                if self.unleash_event_callback and feature.impression_data:
                    try:
                        event = UnleashEvent(
                            event_type=UnleashEventType.VARIANT,
                            event_id=uuid.uuid4(),
                            context=context,
                            enabled=variant_check["enabled"],
                            feature_name=feature_name,
                            variant=variant_check["name"],
                        )

                        self.unleash_event_callback(event)
                    except Exception as excep:
                        LOGGER.log(
                            self.unleash_verbose_log_level,
                            "Error in event callback: %s",
                            excep,
                        )
                        return variant_check

                return variant_check
            except Exception as excep:
                LOGGER.log(
                    self.unleash_verbose_log_level,
                    "Returning default flag/variation for feature: %s",
                    feature_name,
                )
                LOGGER.log(
                    self.unleash_verbose_log_level,
                    "Error checking feature flag variant: %s",
                    excep,
                )

                # The feature doesn't exist, so create it to track metrics
                new_feature = Feature.metrics_only_feature(feature_name)
                self.features[feature_name] = new_feature

                # Use the feature's get_variant method to count the call
                variant_check = new_feature.get_variant(context)
                return variant_check
        else:
            LOGGER.log(
                self.unleash_verbose_log_level,
                "Returning default flag/variation for feature: %s",
                feature_name,
            )
            LOGGER.log(
                self.unleash_verbose_log_level,
                "Attempted to get feature flag/variation %s, but client wasn't initialized!",
                feature_name,
            )
            return disabled_variation

    def _is_dependency_satified(self, dependency: dict, context: dict) -> bool:
        """
        Checks a single feature dependency.
        """

        dependency_name = dependency["feature"]

        dependency_feature = self.features[dependency_name]

        if not dependency_feature:
            LOGGER.warning("Feature dependency not found. %s", dependency_name)
            return False

        if dependency_feature.dependencies:
            LOGGER.warning(
                "Feature dependency cannot have it's own dependencies. %s",
                dependency_name,
            )
            return False

        should_be_enabled = dependency.get("enabled", True)
        is_enabled = dependency_feature.is_enabled(context, skip_stats=True)

        if is_enabled != should_be_enabled:
            return False

        variants = dependency.get("variants")
        if variants:
            variant = dependency_feature.get_variant(context, skip_stats=True)
            if variant["name"] not in variants:
                return False

        return True

    def _dependencies_are_satisfied(self, feature_name: str, context: dict) -> bool:
        """
        If feature dependencies are satisfied (or non-existent).
        """

        feature = self.features[feature_name]
        dependencies = feature.dependencies

        if not dependencies:
            return True

        for dependency in dependencies:
            if not self._is_dependency_satified(dependency, context):
                return False

        return True

    def _do_instance_check(self, multiple_instance_mode):
        identifier = self.__get_identifier()
        if identifier in INSTANCES:
            msg = f"You already have {INSTANCES.count(identifier)} instance(s) configured for this config: {identifier}, please double check the code where this client is being instantiated."
            if multiple_instance_mode == InstanceAllowType.BLOCK:
                raise Exception(msg)  # pylint: disable=broad-exception-raised
            if multiple_instance_mode == InstanceAllowType.WARN:
                LOGGER.error(msg)
        INSTANCES.increment(identifier)

    def __get_identifier(self):
        api_key = (
            self.unleash_custom_headers.get("Authorization")
            if self.unleash_custom_headers is not None
            else None
        )
        return f"apiKey:{api_key} appName:{self.unleash_app_name} instanceId:{self.unleash_instance_id}"

    def __enter__(self) -> "UnleashClient":
        self.initialize_client()
        return self

    def __exit__(self, *args, **kwargs):
        self.destroy()
        return False

1	# pylint: disable=invalid-name
2	import copy	1✔
3	import random	1✔
4	import string	1✔
5	import uuid	1✔
6	import warnings	1✔
7	from datetime import datetime, timezone	1✔
8	from typing import Callable, Optional	1✔
9
10	from apscheduler.executors.pool import ThreadPoolExecutor	1✔
11	from apscheduler.job import Job	1✔
12	from apscheduler.schedulers.background import BackgroundScheduler	1✔
13	from apscheduler.schedulers.base import BaseScheduler	1✔
14	from apscheduler.triggers.interval import IntervalTrigger	1✔
15
16	from UnleashClient.api import register_client	1✔
17	from UnleashClient.constants import DISABLED_VARIATION, ETAG, METRIC_LAST_SENT_TIME	1✔
18	from UnleashClient.events import UnleashEvent, UnleashEventType	1✔
19	from UnleashClient.features import Feature	1✔
20	from UnleashClient.loader import load_features	1✔
21	from UnleashClient.periodic_tasks import (	1✔
22	aggregate_and_send_metrics,
23	fetch_and_load_features,
24	)
25	from UnleashClient.strategies import (	1✔
26	ApplicationHostname,
27	Default,
28	FlexibleRollout,
29	GradualRolloutRandom,
30	GradualRolloutSessionId,
31	GradualRolloutUserId,
32	RemoteAddress,
33	UserWithId,
34	)
35
36	from .cache import BaseCache, FileCache	1✔
37	from .deprecation_warnings import strategy_v2xx_deprecation_check	1✔
38	from .utils import LOGGER, InstanceAllowType, InstanceCounter	1✔
39
40	INSTANCES = InstanceCounter()	1✔
41
42
43	# pylint: disable=dangerous-default-value
44	class UnleashClient:	1✔
45	"""
46	A client for the Unleash feature toggle system.
47
48	:param url: URL of the unleash server, required.
49	:param app_name: Name of the application using the unleash client, required.
50	:param environment: Name of the environment using the unleash client, optional & defaults to "default".
51	:param instance_id: Unique identifier for unleash client instance, optional & defaults to "unleash-client-python"
52	:param refresh_interval: Provisioning refresh interval in seconds, optional & defaults to 15 seconds
53	:param refresh_jitter: Provisioning refresh interval jitter in seconds, optional & defaults to None
54	:param metrics_interval: Metrics refresh interval in seconds, optional & defaults to 60 seconds
55	:param metrics_jitter: Metrics refresh interval jitter in seconds, optional & defaults to None
56	:param disable_metrics: Disables sending metrics to unleash server, optional & defaults to false.
57	:param disable_registration: Disables registration with unleash server, optional & defaults to false.
58	:param custom_headers: Default headers to send to unleash server, optional & defaults to empty.
59	:param custom_options: Default requests parameters, optional & defaults to empty. Can be used to skip SSL verification.
60	:param custom_strategies: Dictionary of custom strategy names : custom strategy objects.
61	:param cache_directory: Location of the cache directory. When unset, FCache will determine the location.
62	:param verbose_log_level: Numerical log level (https://docs.python.org/3/library/logging.html#logging-levels) for cases where checking a feature flag fails.
63	:param cache: Custom cache implementation that extends UnleashClient.cache.BaseCache. When unset, UnleashClient will use Fcache.
64	:param scheduler: Custom APScheduler object. Use this if you want to customize jobstore or executors. When unset, UnleashClient will create it's own scheduler.
65	:param scheduler_executor: Name of APSCheduler executor to use if using a custom scheduler.
66	:param multiple_instance_mode: Determines how multiple instances being instantiated is handled by the SDK, when set to InstanceAllowType.BLOCK, the client constructor will fail when more than one instance is detected, when set to InstanceAllowType.WARN, multiple instances will be allowed but log a warning, when set to InstanceAllowType.SILENTLY_ALLOW, no warning or failure will be raised when instantiating multiple instances of the client. Defaults to InstanceAllowType.WARN
67	:param event_callback: Function to call if impression events are enabled. WARNING: Depending on your event library, this may have performance implications!
68	"""
69
70	def __init__(	1✔
71	self,
72	url: str,
73	app_name: str,
74	environment: str = "default",
75	instance_id: str = "unleash-client-python",
76	refresh_interval: int = 15,
77	refresh_jitter: Optional[int] = None,
78	metrics_interval: int = 60,
79	metrics_jitter: Optional[int] = None,
80	disable_metrics: bool = False,
81	disable_registration: bool = False,
82	custom_headers: Optional[dict] = None,
83	custom_options: Optional[dict] = None,
84	custom_strategies: Optional[dict] = None,
85	cache_directory: Optional[str] = None,
86	project_name: Optional[str] = None,
87	verbose_log_level: int = 30,
88	cache: Optional[BaseCache] = None,
89	scheduler: Optional[BaseScheduler] = None,
90	scheduler_executor: Optional[str] = None,
91	multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN,
92	event_callback: Optional[Callable[[UnleashEvent], None]] = None,
93	) -> None:
94	custom_headers = custom_headers or {}	1✔
95	custom_options = custom_options or {}	1✔
96	custom_strategies = custom_strategies or {}	1✔
97
98	# Configuration
99	self.unleash_url = url.rstrip("/")	1✔
100	self.unleash_app_name = app_name	1✔
101	self.unleash_environment = environment	1✔
102	self.unleash_instance_id = instance_id	1✔
103	self.unleash_refresh_interval = refresh_interval	1✔
104	self.unleash_refresh_jitter = (	1✔
105	int(refresh_jitter) if refresh_jitter is not None else None
106	)
107	self.unleash_metrics_interval = metrics_interval	1✔
108	self.unleash_metrics_jitter = (	1✔
109	int(metrics_jitter) if metrics_jitter is not None else None
110	)
111	self.unleash_disable_metrics = disable_metrics	1✔
112	self.unleash_disable_registration = disable_registration	1✔
113	self.unleash_custom_headers = custom_headers	1✔
114	self.unleash_custom_options = custom_options	1✔
115	self.unleash_static_context = {	1✔
116	"appName": self.unleash_app_name,
117	"environment": self.unleash_environment,
118	}
119	self.unleash_project_name = project_name	1✔
120	self.unleash_verbose_log_level = verbose_log_level	1✔
121	self.unleash_event_callback = event_callback	1✔
122
123	self._do_instance_check(multiple_instance_mode)	1✔
124
125	# Class objects
126	self.features: dict = {}	1✔
127	self.fl_job: Job = None	1✔
128	self.metric_job: Job = None	1✔
129
130	self.cache = cache or FileCache(	1✔
131	self.unleash_app_name, directory=cache_directory
132	)
133	self.cache.mset({METRIC_LAST_SENT_TIME: datetime.now(timezone.utc), ETAG: ""})	1✔
134	self.unleash_bootstrapped = self.cache.bootstrapped	1✔
135
136	# Scheduler bootstrapping
137	# - Figure out the Unleash executor name.
138	if scheduler and scheduler_executor:	1✔
139	self.unleash_executor_name = scheduler_executor	1✔
140	elif scheduler and not scheduler_executor:	1✔
UNCOV 141	raise ValueError(	×
142	"If using a custom scheduler, you must specify a executor."
143	)
144	else:
145	if not scheduler and scheduler_executor:	1✔
UNCOV 146	LOGGER.warning(	×
147	"scheduler_executor should only be used with a custom scheduler."
148	)
149
150	self.unleash_executor_name = f"unleash_executor_{''.join(random.choices(string.ascii_uppercase + string.digits, k=6))}"	1✔
151
152	# Set up the scheduler.
153	if scheduler:	1✔
154	self.unleash_scheduler = scheduler	1✔
155	else:
156	executors = {self.unleash_executor_name: ThreadPoolExecutor()}	1✔
157	self.unleash_scheduler = BackgroundScheduler(executors=executors)	1✔
158
159	# Mappings
160	default_strategy_mapping = {	1✔
161	"applicationHostname": ApplicationHostname,
162	"default": Default,
163	"gradualRolloutRandom": GradualRolloutRandom,
164	"gradualRolloutSessionId": GradualRolloutSessionId,
165	"gradualRolloutUserId": GradualRolloutUserId,
166	"remoteAddress": RemoteAddress,
167	"userWithId": UserWithId,
168	"flexibleRollout": FlexibleRollout,
169	}
170
171	if custom_strategies:	1✔
172	strategy_v2xx_deprecation_check(	1✔
173	[x for x in custom_strategies.values()]
174	) # pylint: disable=R1721
175
176	self.strategy_mapping = {custom_strategies, default_strategy_mapping}	1✔
177
178	# Client status
179	self.is_initialized = False	1✔
180
181	# Bootstrapping
182	if self.unleash_bootstrapped:	1✔
183	load_features(	1✔
184	cache=self.cache,
185	feature_toggles=self.features,
186	strategy_mapping=self.strategy_mapping,
187	)
188
189	def initialize_client(self, fetch_toggles: bool = True) -> None:	1✔
190	"""
191	Initializes client and starts communication with central unleash server(s).
192
193	This kicks off:
194
195	* Client registration
196	* Provisioning poll
197	* Stats poll
198
199	If `fetch_toggles` is `False`, feature toggle polling will be turned off
200	and instead the client will only load features from the cache. This is
201	usually used to cater the multi-process setups, e.g. Django, Celery,
202	etc.
203
204	This will raise an exception on registration if the URL is invalid. It is done automatically if called inside a context manager as in:
205
206	.. code-block:: python
207
208	with UnleashClient(
209	url="https://foo.bar",
210	app_name="myClient1",
211	instance_id="myinstanceid"
212	) as client:
213	pass
214	"""
215	# Only perform initialization steps if client is not initialized.
216	if not self.is_initialized:	1✔
217	# pylint: disable=no-else-raise
218	try:	1✔
219	# Setup
220	metrics_args = {	1✔
221	"url": self.unleash_url,
222	"app_name": self.unleash_app_name,
223	"instance_id": self.unleash_instance_id,
224	"custom_headers": self.unleash_custom_headers,
225	"custom_options": self.unleash_custom_options,
226	"features": self.features,
227	"cache": self.cache,
228	}
229
230	# Register app
231	if not self.unleash_disable_registration:	1✔
232	register_client(	1✔
233	self.unleash_url,
234	self.unleash_app_name,
235	self.unleash_instance_id,
236	self.unleash_metrics_interval,
237	self.unleash_custom_headers,
238	self.unleash_custom_options,
239	self.strategy_mapping,
240	)
241
242	if fetch_toggles:	1✔
243	job_args = {	1✔
244	"url": self.unleash_url,
245	"app_name": self.unleash_app_name,
246	"instance_id": self.unleash_instance_id,
247	"custom_headers": self.unleash_custom_headers,
248	"custom_options": self.unleash_custom_options,
249	"cache": self.cache,
250	"features": self.features,
251	"strategy_mapping": self.strategy_mapping,
252	"project": self.unleash_project_name,
253	}
254	job_func: Callable = fetch_and_load_features	1✔
255	else:
256	job_args = {	1✔
257	"cache": self.cache,
258	"feature_toggles": self.features,
259	"strategy_mapping": self.strategy_mapping,
260	}
261	job_func = load_features	1✔
262
263	job_func(**job_args) # type: ignore	1✔
264	# Start periodic jobs
265	self.unleash_scheduler.start()	1✔
266	self.fl_job = self.unleash_scheduler.add_job(	1✔
267	job_func,
268	trigger=IntervalTrigger(
269	seconds=int(self.unleash_refresh_interval),
270	jitter=self.unleash_refresh_jitter,
271	),
272	executor=self.unleash_executor_name,
273	kwargs=job_args,
274	)
275
276	if not self.unleash_disable_metrics:	1✔
277	self.metric_job = self.unleash_scheduler.add_job(	1✔
278	aggregate_and_send_metrics,
279	trigger=IntervalTrigger(
280	seconds=int(self.unleash_metrics_interval),
281	jitter=self.unleash_metrics_jitter,
282	),
283	executor=self.unleash_executor_name,
284	kwargs=metrics_args,
285	)
286	except Exception as excep:	1✔
287	# Log exceptions during initialization. is_initialized will remain false.
288	LOGGER.warning(	1✔
289	"Exception during UnleashClient initialization: %s", excep
290	)
291	raise excep	1✔
292	else:
293	# Set is_initialized to true if no exception is encountered.
294	self.is_initialized = True	1✔
295	else:
296	warnings.warn(	1✔
297	"Attempted to initialize an Unleash Client instance that has already been initialized."
298	)
299
300	def destroy(self) -> None:	1✔
301	"""
302	Gracefully shuts down the Unleash client by stopping jobs, stopping the scheduler, and deleting the cache.
303
304	You shouldn't need this too much!
305	"""
306	self.fl_job.remove()	1✔
307	if self.metric_job:	1✔
308	self.metric_job.remove()	1✔
309	self.unleash_scheduler.shutdown()	1✔
310	self.cache.destroy()	1✔
311
312	@staticmethod	1✔
313	def _get_fallback_value(	1✔
314	fallback_function: Callable, feature_name: str, context: dict
315	) -> bool:
316	if fallback_function:	1✔
317	fallback_value = fallback_function(feature_name, context)	1✔
318	else:
319	fallback_value = False	1✔
320
321	return fallback_value	1✔
322
323	# pylint: disable=broad-except
324	def is_enabled(	1✔
325	self,
326	feature_name: str,
327	context: Optional[dict] = None,
328	fallback_function: Callable = None,
329	) -> bool:
330	"""
331	Checks if a feature toggle is enabled.
332
333	Notes:
334
335	* If client hasn't been initialized yet or an error occurs, flat will default to false.
336
337	:param feature_name: Name of the feature
338	:param context: Dictionary with context (e.g. IPs, email) for feature toggle.
339	:param fallback_function: Allows users to provide a custom function to set default value.
340	:return: Feature flag result
341	"""
342	context = context or {}	1✔
343
344	base_context = self.unleash_static_context.copy()	1✔
345	# Update context with static values and allow context to override environment
346	base_context.update(context)	1✔
347	context = base_context	1✔
348
349	if self.unleash_bootstrapped or self.is_initialized:	1✔
350	try:	1✔
351	feature = self.features[feature_name]	1✔
352	feature_check = feature.is_enabled(	1✔
353	context
354	) and self._dependencies_are_satisfied(feature_name, context)
355
356	if feature.only_for_metrics:	1✔
357	return self._get_fallback_value(	1✔
358	fallback_function, feature_name, context
359	)
360
361	try:	1✔
362	if self.unleash_event_callback and feature.impression_data:	1✔
363	event = UnleashEvent(	1✔
364	event_type=UnleashEventType.FEATURE_FLAG,
365	event_id=uuid.uuid4(),
366	context=context,
367	enabled=feature_check,
368	feature_name=feature_name,
369	)
370
371	self.unleash_event_callback(event)	1✔
372	except Exception as excep:	1✔
373	LOGGER.log(	1✔
374	self.unleash_verbose_log_level,
375	"Error in event callback: %s",
376	excep,
377	)
378	return feature_check	1✔
379
380	return feature_check	1✔
381	except Exception as excep:	1✔
382	LOGGER.log(	1✔
383	self.unleash_verbose_log_level,
384	"Returning default value for feature: %s",
385	feature_name,
386	)
387	LOGGER.log(	1✔
388	self.unleash_verbose_log_level,
389	"Error checking feature flag: %s",
390	excep,
391	)
392	# The feature doesn't exist, so create it to track metrics
393	new_feature = Feature.metrics_only_feature(feature_name)	1✔
394	self.features[feature_name] = new_feature	1✔
395
396	# Use the feature's is_enabled method to count the call
397	new_feature.is_enabled(context)	1✔
398
399	return self._get_fallback_value(	1✔
400	fallback_function, feature_name, context
401	)
402
403	else:
404	LOGGER.log(	1✔
405	self.unleash_verbose_log_level,
406	"Returning default value for feature: %s",
407	feature_name,
408	)
409	LOGGER.log(	1✔
410	self.unleash_verbose_log_level,
411	"Attempted to get feature_flag %s, but client wasn't initialized!",
412	feature_name,
413	)
414	return self._get_fallback_value(fallback_function, feature_name, context)	1✔
415
416	# pylint: disable=broad-except
417	def get_variant(self, feature_name: str, context: Optional[dict] = None) -> dict:	1✔
418	"""
419	Checks if a feature toggle is enabled. If so, return variant.
420
421	Notes:
422
423	* If client hasn't been initialized yet or an error occurs, flat will default to false.
424
425	:param feature_name: Name of the feature
426	:param context: Dictionary with context (e.g. IPs, email) for feature toggle.
427	:return: Variant and feature flag status.
428	"""
429	context = context or {}	1✔
430	context.update(self.unleash_static_context)	1✔
431	disabled_variation = copy.deepcopy(DISABLED_VARIATION)	1✔
432
433	if self.unleash_bootstrapped or self.is_initialized:	1✔
434	try:	1✔
435	feature = self.features[feature_name]	1✔
436
437	if not self._dependencies_are_satisfied(feature_name, context):	1✔
NEW UNCOV 438	return disabled_variation	×
439
440	variant_check = feature.get_variant(context)	1✔
441
442	if self.unleash_event_callback and feature.impression_data:	1✔
443	try:	1✔
444	event = UnleashEvent(	1✔
445	event_type=UnleashEventType.VARIANT,
446	event_id=uuid.uuid4(),
447	context=context,
448	enabled=variant_check["enabled"],
449	feature_name=feature_name,
450	variant=variant_check["name"],
451	)
452
453	self.unleash_event_callback(event)	1✔
454	except Exception as excep:	1✔
455	LOGGER.log(	1✔
456	self.unleash_verbose_log_level,
457	"Error in event callback: %s",
458	excep,
459	)
460	return variant_check	1✔
461
462	return variant_check	1✔
463	except Exception as excep:	1✔
464	LOGGER.log(	1✔
465	self.unleash_verbose_log_level,
466	"Returning default flag/variation for feature: %s",
467	feature_name,
468	)
469	LOGGER.log(	1✔
470	self.unleash_verbose_log_level,
471	"Error checking feature flag variant: %s",
472	excep,
473	)
474
475	# The feature doesn't exist, so create it to track metrics
476	new_feature = Feature.metrics_only_feature(feature_name)	1✔
477	self.features[feature_name] = new_feature	1✔
478
479	# Use the feature's get_variant method to count the call
480	variant_check = new_feature.get_variant(context)	1✔
481	return variant_check	1✔
482	else:
483	LOGGER.log(	1✔
484	self.unleash_verbose_log_level,
485	"Returning default flag/variation for feature: %s",
486	feature_name,
487	)
488	LOGGER.log(	1✔
489	self.unleash_verbose_log_level,
490	"Attempted to get feature flag/variation %s, but client wasn't initialized!",
491	feature_name,
492	)
493	return disabled_variation	1✔
494
495	def _is_dependency_satified(self, dependency: dict, context: dict) -> bool:	1✔
496	"""
497	Checks a single feature dependency.
498	"""
499
500	dependency_name = dependency["feature"]	1✔
501
502	dependency_feature = self.features[dependency_name]	1✔
503
504	if not dependency_feature:	1✔
UNCOV 505	LOGGER.warning("Feature dependency not found. %s", dependency_name)	×
UNCOV 506	return False	×
507
508	if dependency_feature.dependencies:	1✔
509	LOGGER.warning(	1✔
510	"Feature dependency cannot have it's own dependencies. %s",
511	dependency_name,
512	)
513	return False	1✔
514
515	should_be_enabled = dependency.get("enabled", True)	1✔
516	is_enabled = dependency_feature.is_enabled(context, skip_stats=True)	1✔
517
518	if is_enabled != should_be_enabled:	1✔
519	return False	1✔
520
521	variants = dependency.get("variants")	1✔
522	if variants:	1✔
523	variant = dependency_feature.get_variant(context, skip_stats=True)	1✔
524	if variant["name"] not in variants:	1✔
UNCOV 525	return False	×
526
527	return True	1✔
528
529	def _dependencies_are_satisfied(self, feature_name: str, context: dict) -> bool:	1✔
530	"""
531	If feature dependencies are satisfied (or non-existent).
532	"""
533
534	feature = self.features[feature_name]	1✔
535	dependencies = feature.dependencies	1✔
536
537	if not dependencies:	1✔
538	return True	1✔
539
540	for dependency in dependencies:	1✔
541	if not self._is_dependency_satified(dependency, context):	1✔
542	return False	1✔
543
544	return True	1✔
545
546	def _do_instance_check(self, multiple_instance_mode):	1✔
547	identifier = self.__get_identifier()	1✔
548	if identifier in INSTANCES:	1✔
549	msg = f"You already have {INSTANCES.count(identifier)} instance(s) configured for this config: {identifier}, please double check the code where this client is being instantiated."	1✔
550	if multiple_instance_mode == InstanceAllowType.BLOCK:	1✔
551	raise Exception(msg) # pylint: disable=broad-exception-raised	1✔
552	if multiple_instance_mode == InstanceAllowType.WARN:	1✔
553	LOGGER.error(msg)	1✔
554	INSTANCES.increment(identifier)	1✔
555
556	def __get_identifier(self):	1✔
557	api_key = (	1✔
558	self.unleash_custom_headers.get("Authorization")
559	if self.unleash_custom_headers is not None
560	else None
561	)
562	return f"apiKey:{api_key} appName:{self.unleash_app_name} instanceId:{self.unleash_instance_id}"	1✔
563
564	def __enter__(self) -> "UnleashClient":	1✔
565	self.initialize_client()	1✔
566	return self	1✔
567
568	def __exit__(self, args, *kwargs):	1✔
569	self.destroy()	1✔
570	return False	1✔

Unleash / unleash-client-python / 6782905699

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous