[FSSDK-11184] Update: Send CMAB uuid in impression events

optimizely/bucketer.py

@@ -119,6 +119,34 @@ def bucket(
             and array of log messages representing decision making.
      */.
         """
+        variation_id, decide_reasons = self.bucket_to_entity_id(project_config, experiment, user_id, bucketing_id)
+        if variation_id:
+            variation = project_config.get_variation_from_id_by_experiment_id(experiment.id, variation_id)
+            return variation, decide_reasons
+
+        elif not decide_reasons:
+            message = 'Bucketed into an empty traffic range. Returning nil.'
+            project_config.logger.info(message)
+            decide_reasons.append(message)
+
+        return None, decide_reasons
+
+    def bucket_to_entity_id(
+        self, project_config: ProjectConfig,
+        experiment: Experiment, user_id: str, bucketing_id: str
+    ) -> tuple[Optional[str], list[str]]:
+        """
+        For a given experiment and bucketing ID determines variation ID to be shown to user.
+
+        Args:
+            project_config: Instance of ProjectConfig.
+            experiment: The experiment object (used for group/groupPolicy logic if needed).
+            user_id: The user ID string.
+            bucketing_id: The bucketing ID string for the user.
+
+        Returns:
+            Tuple of (entity_id or None, list of decide reasons).
+        """
         decide_reasons: list[str] = []
         if not experiment:
             return None, decide_reasons
@@ -151,16 +179,16 @@ def bucket(
             project_config.logger.info(message)
             decide_reasons.append(message)
 
+        traffic_allocations: list[TrafficAllocation] = experiment.trafficAllocation
+        if experiment.cmab:
+            traffic_allocations = [
+                {
+                    "entityId": "$",
+                    "endOfRange": experiment.cmab['trafficAllocation']
+                }
+            ]
         # Bucket user if not in white-list and in group (if any)
         variation_id = self.find_bucket(project_config, bucketing_id,
-                                        experiment.id, experiment.trafficAllocation)
-        if variation_id:
-            variation = project_config.get_variation_from_id_by_experiment_id(experiment.id, variation_id)
-            return variation, decide_reasons
-
-        else:
-            message = 'Bucketed into an empty traffic range. Returning nil.'
-            project_config.logger.info(message)
-            decide_reasons.append(message)
+                                        experiment.id, traffic_allocations)
 
-        return None, decide_reasons
+        return variation_id, decide_reasons

optimizely/cmab/__init__.py

@@ -0,0 +1,12 @@
+# Copyright 2025, Optimizely
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

optimizely/cmab/cmab_service.py

@@ -35,6 +35,18 @@ class CmabCacheValue(TypedDict):
 
 
 class DefaultCmabService:
+    """
+    DefaultCmabService handles decisioning for Contextual Multi-Armed Bandit (CMAB) experiments,
+    including caching and filtering user attributes for efficient decision retrieval.
+
+    Attributes:
+        cmab_cache: LRUCache for user CMAB decisions.
+        cmab_client: Client to fetch decisions from the CMAB backend.
+        logger: Optional logger.
+
+    Methods:
+        get_decision: Retrieves a CMAB decision with caching and attribute filtering.
+    """
     def __init__(self, cmab_cache: LRUCache[str, CmabCacheValue],
                  cmab_client: DefaultCmabClient, logger: Optional[_logging.Logger] = None):
         self.cmab_cache = cmab_cache

optimizely/decision/optimizely_decision.py

@@ -48,3 +48,23 @@ def as_json(self) -> dict[str, Any]:
             'user_context': self.user_context.as_json() if self.user_context else None,
             'reasons': self.reasons
         }
+
+    @classmethod
+    def new_error_decision(cls, key: str, user: OptimizelyUserContext, reasons: list[str]) -> OptimizelyDecision:
+        """Create a new OptimizelyDecision representing an error state.
+        Args:
+            key: The flag key
+            user: The user context
+            reasons: List of reasons explaining the error
+        Returns:
+            OptimizelyDecision with error state values
+        """
+        return cls(
+            variation_key=None,
+            enabled=False,
+            variables={},
+            rule_key=None,
+            flag_key=key,
+            user_context=user,
+            reasons=[reasons[-1]] if reasons else []
+        )