Added Accessor

featuremanagement/_featuremanager.py

@@ -3,7 +3,8 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
-from typing import cast, overload, Any, Optional, Dict, Mapping, List
+import logging
+from typing import cast, overload, Any, Optional, Dict, Mapping, List, Tuple
 from ._defaultfilters import TimeWindowFilter, TargetingFilter
 from ._featurefilters import FeatureFilter
 from ._models import EvaluationEvent, Variant, TargetingContext
@@ -14,6 +15,8 @@
     FEATURE_FILTER_NAME,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class FeatureManager(FeatureManagerBase):
     """
@@ -23,6 +26,8 @@ class FeatureManager(FeatureManagerBase):
     :keyword list[FeatureFilter] feature_filters: Custom filters to be used for evaluating feature flags.
     :keyword Callable[EvaluationEvent] on_feature_evaluated: Callback function to be called when a feature flag is
     evaluated.
+    :keyword Callable[[], TargetingContext] targeting_context_accessor: Callback function to get the current targeting
+    context if one isn't provided.
     """
 
     def __init__(self, configuration: Mapping[str, Any], **kwargs: Any):
@@ -56,7 +61,7 @@ def is_enabled(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> bool:
         :return: True if the feature flag is enabled for the given context.
         :rtype: bool
         """
-        targeting_context = self._build_targeting_context(args)
+        targeting_context: TargetingContext = self._build_targeting_context(args)
 
         result = self._check_feature(feature_flag_id, targeting_context, **kwargs)
         if (
@@ -89,7 +94,7 @@ def get_variant(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> Option
         :return: Variant instance.
         :rtype: Variant
         """
-        targeting_context = self._build_targeting_context(args)
+        targeting_context: TargetingContext = self._build_targeting_context(args)
 
         result = self._check_feature(feature_flag_id, targeting_context, **kwargs)
         if (
@@ -102,6 +107,21 @@ def get_variant(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> Option
             self._on_feature_evaluated(result)
         return result.variant
 
+    def _build_targeting_context(self, args: Tuple[Any]) -> TargetingContext:
+        targeting_context = super()._build_targeting_context(args)
+        if targeting_context:
+            return targeting_context
+        if not targeting_context and self._targeting_context_accessor and callable(self._targeting_context_accessor):
+            targeting_context = self._targeting_context_accessor()
+            if targeting_context and isinstance(targeting_context, TargetingContext):
+                return targeting_context
+            logger.warning(
+                "targeting_context_accessor did not return a TargetingContext. Received type %s.",
+                type(targeting_context),
+            )
+
+        return TargetingContext()
+
     def _check_feature_filters(
         self, evaluation_event: EvaluationEvent, targeting_context: TargetingContext, **kwargs: Any
     ) -> None:

featuremanagement/_featuremanagerbase.py

@@ -6,7 +6,7 @@
 import hashlib
 import logging
 from abc import ABC
-from typing import List, Optional, Dict, Tuple, Any, Mapping
+from typing import List, Optional, Dict, Tuple, Any, Mapping, Callable
 from ._models import FeatureFlag, Variant, VariantAssignmentReason, TargetingContext, EvaluationEvent, VariantReference
 
 
@@ -21,6 +21,9 @@
 FEATURE_FILTER_PARAMETERS = "parameters"
 
 
+logger = logging.getLogger(__name__)
+
+
 def _get_feature_flag(configuration: Mapping[str, Any], feature_flag_name: str) -> Optional[FeatureFlag]:
     """
     Gets the FeatureFlag json from the configuration, if it exists it gets converted to a FeatureFlag object.
@@ -77,6 +80,9 @@ def __init__(self, configuration: Mapping[str, Any], **kwargs: Any):
         self._cache: Dict[str, Optional[FeatureFlag]] = {}
         self._copy = configuration.get(FEATURE_MANAGEMENT_KEY)
         self._on_feature_evaluated = kwargs.pop("on_feature_evaluated", None)
+        self._targeting_context_accessor: Optional[Callable[[], TargetingContext]] = kwargs.pop(
+            "targeting_context_accessor", None
+        )
 
     @staticmethod
     def _assign_default_disabled_variant(evaluation_event: EvaluationEvent) -> None:
@@ -218,7 +224,7 @@ def _variant_name_to_variant(self, feature_flag: FeatureFlag, variant_name: Opti
                 return Variant(variant_reference.name, variant_reference.configuration_value)
         return None
 
-    def _build_targeting_context(self, args: Tuple[Any]) -> TargetingContext:
+    def _build_targeting_context(self, args: Tuple[Any]) -> Optional[TargetingContext]:
         """
         Builds a TargetingContext, either returns a provided context, takes the provided user_id to make a context, or
         returns an empty context.
@@ -229,10 +235,12 @@ def _build_targeting_context(self, args: Tuple[Any]) -> TargetingContext:
         if len(args) == 1:
             arg = args[0]
             if isinstance(arg, str):
+                # If the user_id is provided, return a TargetingContext with the user_id
                 return TargetingContext(user_id=arg, groups=[])
             if isinstance(arg, TargetingContext):
+                # If a TargetingContext is provided, return it
                 return arg
-        return TargetingContext()
+        return None
 
     def _assign_allocation(self, evaluation_event: EvaluationEvent, targeting_context: TargetingContext) -> None:
         feature_flag = evaluation_event.feature
@@ -271,7 +279,7 @@ def _check_feature_base(self, feature_flag_id: str) -> Tuple[EvaluationEvent, bo
 
         evaluation_event = EvaluationEvent(feature_flag)
         if not feature_flag:
-            logging.warning("Feature flag %s not found", feature_flag_id)
+            logger.warning("Feature flag %s not found", feature_flag_id)
             # Unknown feature flags are disabled by default
             return evaluation_event, True
 

featuremanagement/aio/_featuremanager.py

@@ -4,7 +4,8 @@
 # license information.
 # -------------------------------------------------------------------------
 import inspect
-from typing import cast, overload, Any, Optional, Dict, Mapping, List
+import logging
+from typing import cast, overload, Any, Optional, Dict, Mapping, List, Tuple
 from ._defaultfilters import TimeWindowFilter, TargetingFilter
 from ._featurefilters import FeatureFilter
 from .._models import EvaluationEvent, Variant, TargetingContext
@@ -15,6 +16,8 @@
     FEATURE_FILTER_NAME,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class FeatureManager(FeatureManagerBase):
     """
@@ -24,6 +27,8 @@ class FeatureManager(FeatureManagerBase):
     :keyword list[FeatureFilter] feature_filters: Custom filters to be used for evaluating feature flags.
     :keyword Callable[EvaluationEvent] on_feature_evaluated: Callback function to be called when a feature flag is
     evaluated.
+    :keyword Callable[[], TargetingContext] targeting_context_accessor: Callback function to get the current targeting
+    context if one isn't provided.
     """
 
     def __init__(self, configuration: Mapping[str, Any], **kwargs: Any):
@@ -57,7 +62,7 @@ async def is_enabled(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> b
         :return: True if the feature flag is enabled for the given context.
         :rtype: bool
         """
-        targeting_context = self._build_targeting_context(args)
+        targeting_context: TargetingContext = await self._build_targeting_context_async(args)
 
         result = await self._check_feature(feature_flag_id, targeting_context, **kwargs)
         if (
@@ -93,7 +98,7 @@ async def get_variant(self, feature_flag_id: str, *args: Any, **kwargs: Any) ->
         :return: Variant instance.
         :rtype: Variant
         """
-        targeting_context = self._build_targeting_context(args)
+        targeting_context: TargetingContext = await self._build_targeting_context_async(args)
 
         result = await self._check_feature(feature_flag_id, targeting_context, **kwargs)
         if (
@@ -109,6 +114,25 @@ async def get_variant(self, feature_flag_id: str, *args: Any, **kwargs: Any) ->
                 self._on_feature_evaluated(result)
         return result.variant
 
+    async def _build_targeting_context_async(self, args: Tuple[Any]) -> TargetingContext:
+        targeting_context = super()._build_targeting_context(args)
+        if targeting_context:
+            return targeting_context
+        if not targeting_context and self._targeting_context_accessor and callable(self._targeting_context_accessor):
+
+            if inspect.iscoroutinefunction(self._targeting_context_accessor):
+                # If a targeting_context_accessor is provided, return the TargetingContext from it
+                targeting_context = await self._targeting_context_accessor()
+            else:
+                targeting_context = self._targeting_context_accessor()
+            if targeting_context and isinstance(targeting_context, TargetingContext):
+                return targeting_context
+            logger.warning(
+                "targeting_context_accessor did not return a TargetingContext. Received type %s.",
+                type(targeting_context),
+            )
+        return TargetingContext()
+
     async def _check_feature_filters(
         self, evaluation_event: EvaluationEvent, targeting_context: TargetingContext, **kwargs: Any
     ) -> None:

samples/feature_variant_sample_with_targeting_accessor.py

@@ -0,0 +1,36 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+import json
+import os
+import sys
+from random_filter import RandomFilter
+from featuremanagement import FeatureManager, TargetingContext
+
+
+script_directory = os.path.dirname(os.path.abspath(sys.argv[0]))
+
+with open(script_directory + "/formatted_feature_flags.json", "r", encoding="utf-8") as f:
+    feature_flags = json.load(f)
+
+USER_ID = "Adam"
+
+
+def my_targeting_accessor() -> TargetingContext:
+    return TargetingContext(user_id=USER_ID)
+
+
+feature_manager = FeatureManager(
+    feature_flags, feature_filters=[RandomFilter()], targeting_context_accessor=my_targeting_accessor
+)
+
+print(feature_manager.is_enabled("TestVariants"))
+print(feature_manager.get_variant("TestVariants").configuration)
+
+USER_ID = "Ellie"
+
+print(feature_manager.is_enabled("TestVariants"))
+print(feature_manager.get_variant("TestVariants").configuration)

samples/formatted_feature_flags.json

@@ -218,6 +218,7 @@
                     },
                     {
                         "name": "False_Override",
+                        "configuration_value": "The Variant False_Override overrides to True",
                         "status_override": "True"
                     }
                 ]

tests/test_default_feature_flags.py

@@ -261,3 +261,52 @@ def test_feature_manager_requirement_type(self):
         # The second TimeWindow filter failed
         assert not feature_manager.is_enabled("Beta")
         assert feature_manager.is_enabled("Gamma")
+
+    def test_feature_manager_with_targeting_accessor(self):
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "Target",
+                        "enabled": "true",
+                        "conditions": {
+                            "client_filters": [
+                                {
+                                    "name": "Microsoft.Targeting",
+                                    "parameters": {
+                                        "Audience": {
+                                            "Users": ["Adam"],
+                                            "Groups": [{"Name": "Stage1", "RolloutPercentage": 100}],
+                                            "DefaultRolloutPercentage": 50,
+                                            "Exclusion": {"Users": [], "Groups": []},
+                                        }
+                                    },
+                                }
+                            ]
+                        },
+                    },
+                ]
+            }
+        }
+
+        user_id = "Adam"
+        group_id = None
+
+        def my_targeting_accessor() -> TargetingContext:
+            if group_id:
+                return TargetingContext(user_id=user_id, groups=[group_id])
+            return TargetingContext(user_id=user_id)
+
+        feature_manager = FeatureManager(feature_flags, targeting_context_accessor=my_targeting_accessor)
+        assert feature_manager is not None
+        # Adam is in the user audience
+        assert feature_manager.is_enabled("Target")
+        # Belle is not part of the 50% or default 50% of users
+        user_id = "Belle"
+        assert not feature_manager.is_enabled("Target")
+        # Belle is enabled because all of Stage 1 is enabled
+        group_id = "Stage1"
+        assert feature_manager.is_enabled("Target")
+        # Belle is not enabled because he is not in Stage 2, group isn't looked at when user is targeted
+        group_id = "Stage2"
+        assert not feature_manager.is_enabled("Target")

tests/test_default_feature_flags_async.py

@@ -266,3 +266,53 @@ async def test_feature_manager_requirement_type(self):
         # The second TimeWindow filter failed
         assert not await feature_manager.is_enabled("Beta")
         assert await feature_manager.is_enabled("Gamma")
+
+    @pytest.mark.asyncio
+    async def test_feature_manager_with_targeting_accessor(self):
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "Target",
+                        "enabled": "true",
+                        "conditions": {
+                            "client_filters": [
+                                {
+                                    "name": "Microsoft.Targeting",
+                                    "parameters": {
+                                        "Audience": {
+                                            "Users": ["Adam"],
+                                            "Groups": [{"Name": "Stage1", "RolloutPercentage": 100}],
+                                            "DefaultRolloutPercentage": 50,
+                                            "Exclusion": {"Users": [], "Groups": []},
+                                        }
+                                    },
+                                }
+                            ]
+                        },
+                    },
+                ]
+            }
+        }
+
+        user_id = "Adam"
+        group_id = None
+
+        def my_targeting_accessor() -> TargetingContext:
+            if group_id:
+                return TargetingContext(user_id=user_id, groups=[group_id])
+            return TargetingContext(user_id=user_id)
+
+        feature_manager = FeatureManager(feature_flags, targeting_context_accessor=my_targeting_accessor)
+        assert feature_manager is not None
+        # Adam is in the user audience
+        assert await feature_manager.is_enabled("Target")
+        # Belle is not part of the 50% or default 50% of users
+        user_id = "Belle"
+        assert not await feature_manager.is_enabled("Target")
+        # Belle is enabled because all of Stage 1 is enabled
+        group_id = "Stage1"
+        assert await feature_manager.is_enabled("Target")
+        # Belle is not enabled because he is not in Stage 2, group isn't looked at when user is targeted
+        group_id = "Stage2"
+        assert not await feature_manager.is_enabled("Target")