Function _rerank_items

src/personalization_api_function/main.py:447–504

(namespace: str, recommender: str, user_id: str, item_ids: List[str], background: BackgroundTasks)

        )

def _rerank_items(namespace: str, recommender: str, user_id: str, item_ids: List[str], background: BackgroundTasks) -> Tuple[Response, Dict]:
    rec_config = config.get_recommender_config(namespace, recommender, ACTION_RERANK_ITEMS)
    if not rec_config:
        raise ConfigError(HTTPStatus.NOT_FOUND, 'RecommenderNotConfigured', 'Recommender not configured for this namespace and recommender path')

    variation, experiment = evaluate_variations(rec_config, user_id, background)
    context = resolve_context(variation)

    if variation.get('type') in ['personalize-campaign', 'personalize-recommender']:
        arn = variation.get('arn')
        filter_arn, filter_values = resolve_filter_parameters(variation, user_id)

        response = personalize_resolver.rerank_items(
                variation_config = variation,
                arn = arn,
                user_id = user_id,
                input_list = item_ids,
                filter_arn = filter_arn,
                filter_values = filter_values,
                context = context,
                include_metadata = try_decorate_items()
        )

        post_decorate_items(namespace, response)
    elif variation.get('type') == 'sagemaker':
        response = sagemaker_resolver.rerank_items(
                recommender,
                rec_config,
                variation,
                user_id = user_id,
                input_list = item_ids,
                context = context
        )

        post_decorate_items(namespace, response)
    elif variation.get('type') == 'lambda':
        response = lambda_resolver.rerank_items(
                recommender,
                rec_config,
                variation,
                user_id = user_id,
                input_list = item_ids,
                context = context
        )

        post_decorate_items(namespace, response)
    elif variation.get('type') == 'http':
        url = variation['url'].format(**app.current_event.query_string_parameters)
        response = json.loads(urllib.request.urlopen(url).read())

    if response and experiment:
        response['matchedExperiment'] = experiment

    post_process_config = rec_config.get('responsePostProcessor')
    if post_process_config:
        response = post_processor.process_rerank_items(recommender, rec_config, variation, user_id, response)

    return response, variation