6978349191

Committed 24 Nov 2023 08:14AM UTC coverage: 96.898%. Remained the same

Build # 6978349191

Build Type

push

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

Run Details

2 of 2 new or added lines in 2 files 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.07

/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,
    REQUEST_RETRIES,
    REQUEST_TIMEOUT,
)
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
    :params request_timeout: Timeout for requests to unleash server in seconds, optional & defaults to 30 seconds
    :params request_retries: Number of retries for requests to unleash server, optional & defaults to 3
    :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,
        request_timeout: int = REQUEST_TIMEOUT,
        request_retries: int = REQUEST_RETRIES,
        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_request_timeout = request_timeout
        self.unleash_request_retries = request_retries
        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,
                    "request_timeout": self.unleash_request_timeout,
                }

                # 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,
                        self.unleash_request_timeout,
                    )

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

Unleash / unleash-client-python / 6978349191

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