def get_items()

in src/recommendations/src/recommendations-service/experimentation/experiment_interleaving.py [0:0]

• 48 lines of code
• 13 McCabe index (conditional complexity)

    def get_items(self, user_id, current_item_id=None, item_list=None, num_results=10, tracker=None, context=None):
        if not user_id:
            raise Exception('user_id is required')
        if len(self.variations) < 2:
            raise Exception(f'Experiment {self.id} does not have 2 or more variations')

        # Initialize array structure to hold item recommendations for each variation
        variations_data = [[] for x in range(len(self.variations))]

        resolve_params = {
            'user_id': user_id,
            'product_id': current_item_id,
            'product_list': item_list,
            'num_results': num_results * 3,  # account for overlaps
            'context': context
        }

        # Get recomended items for each variation
        for i in range(len(self.variations)):
            variation = self.variations[i]
            items = variation.resolver.get_items(**resolve_params)
            variations_data[i] = items

        # Interleave items to produce result
        interleaved = []
        if self.method == InterleavingExperiment.METHOD_TEAM_DRAFT:
            interleaved = self._interleave_team_draft(user_id, variations_data, num_results)
        else:
            interleaved = self._interleave_balanced(user_id, variations_data, num_results)

        # Increment exposure for each variation (can be optimized)
        for i in range(len(self.variations)):
            self._increment_exposure_count(i)

        if tracker is not None:
            # Track exposure details
            track_interleaved = []
            for item in interleaved:
                track_interleaved.append({
                    'item_id': item['itemId'],
                    'variation_index': item['experiment']['variationIndex']
                })

            event = {
                'event_type': 'Experiment Exposure',
                'event_timestamp': int(round(time.time() * 1000)),
                'attributes': {
                    'user_id': user_id,
                    'experiment': {
                        'id': self.id,
                        'feature': self.feature,
                        'name': self.name,
                        'type': self.type,
                        'method': self.method
                    },
                    'interleaved': track_interleaved
                }
            }

            tracker.log_exposure(event)

        return interleaved