7031328308

Committed 29 Nov 2023 10:35AM UTC coverage: 96.898%. Remained the same

Build # 7031328308

Build Type

Pull #294

github

Committed by

pre-commit-ci[bot]

Commit Message

[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Pull Request Pull Request #294: Fix/double execution result

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

6 existing lines in 1 file now uncovered.

906 of 935 relevant lines covered (96.9%)

0.97 hits per line

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

97.04

/UnleashClient/__init__.py

# pylint: disable=invalid-name
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)

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

Unleash / unleash-client-python / 7031328308

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