Function _augment_response_with_on_demand_transforms

sdk/python/feast/utils.py:726–887

Computes on demand feature values and adds them to the result rows. Assumes that 'online_features_response' already contains the necessary request data and input feature views for the on demand feature views. Unneeded feature values such as request data and unrequested input feature vie

(
    online_features_response: GetOnlineFeaturesResponse,
    feature_refs: List[str],
    requested_on_demand_feature_views: List["OnDemandFeatureView"],
    full_feature_names: bool,
    feature_types: Optional[Dict[str, "ValueType"]] = None,
)

def _augment_response_with_on_demand_transforms(
    online_features_response: GetOnlineFeaturesResponse,
    feature_refs: List[str],
    requested_on_demand_feature_views: List["OnDemandFeatureView"],
    full_feature_names: bool,
    feature_types: Optional[Dict[str, "ValueType"]] = None,
):
    """Computes on demand feature values and adds them to the result rows.

    Assumes that 'online_features_response' already contains the necessary request data and input feature
    views for the on demand feature views. Unneeded feature values such as request data and
    unrequested input feature views will be removed from 'online_features_response'.

    Args:
        online_features_response: Protobuf object to populate
        feature_refs: List of all feature references to be returned.
        requested_on_demand_feature_views: List of all odfvs that have been requested.
        full_feature_names: A boolean that provides the option to add the feature view prefixes to the feature names,
            changing them from the format "feature" to "feature_view__feature" (e.g., "daily_transactions" changes to
            "customer_fv__daily_transactions").
    """
    from feast.online_response import OnlineResponse

    requested_odfv_map = {odfv.name: odfv for odfv in requested_on_demand_feature_views}
    requested_odfv_feature_names = requested_odfv_map.keys()

    odfv_feature_refs = defaultdict(list)
    for feature_ref in feature_refs:
        view_name, _, feature_name = _parse_feature_ref(feature_ref)
        if view_name in requested_odfv_feature_names:
            odfv_feature_refs[view_name].append(
                f"{requested_odfv_map[view_name].projection.name_to_use()}__{feature_name}"
                if full_feature_names
                else feature_name
            )

    initial_response = OnlineResponse(
        online_features_response, feature_types=feature_types
    )
    initial_response_arrow: Optional[pyarrow.Table] = None
    initial_response_dict: Optional[Dict[str, List[Any]]] = None

    def _is_metrics_active():
        try:
            from feast.metrics import _config

            return _config.online_features
        except Exception:
            return False

    _metrics_active = _is_metrics_active()

    # Apply on demand transformations and augment the result rows
    odfv_result_names = set()
    for odfv_name, _feature_refs in odfv_feature_refs.items():
        odfv = requested_odfv_map[odfv_name]
        if not odfv.write_to_online_store:
            _should_track = _metrics_active and getattr(odfv, "track_metrics", False)