Function _get_feature_views_to_use

sdk/python/feast/utils.py:1196–1307

(
    registry: "BaseRegistry",
    project,
    features: Union[List[str], "FeatureService"],
    allow_cache=False,
    hide_dummy_entity: bool = True,
)

def _get_feature_views_to_use(
    registry: "BaseRegistry",
    project,
    features: Union[List[str], "FeatureService"],
    allow_cache=False,
    hide_dummy_entity: bool = True,
) -> Tuple[List["FeatureView"], List["OnDemandFeatureView"]]:
    from feast.feature_service import FeatureService
    from feast.feature_view import DUMMY_ENTITY_NAME
    from feast.on_demand_feature_view import OnDemandFeatureView

    if isinstance(features, FeatureService):
        feature_views = [
            (projection.name, None, projection)
            for projection in features.feature_view_projections
        ]
    else:
        assert features is not None
        # Parse version-qualified refs: 'fv@v2:feat' -> ('fv', 2, None)
        parsed = []
        seen = set()
        for feature in features:
            fv_name, version_num, _ = _parse_feature_ref(feature)
            key = (fv_name, version_num)
            if key not in seen:
                seen.add(key)
                parsed.append((fv_name, version_num, None))
        feature_views = parsed  # type: ignore[assignment]

    fvs_to_use, od_fvs_to_use = [], []
    for name, version_num, projection in feature_views:
        if version_num is not None:
            if not getattr(registry, "enable_online_versioning", False):
                raise ValueError(
                    f"Version-qualified ref '{name}@v{version_num}' not supported: "
                    f"online versioning is disabled. Set 'enable_online_feature_view_versioning: true' "
                    f"under 'registry' in feature_store.yaml."
                )
            # Version-qualified reference: look up the specific version snapshot
            try:
                fv = registry.get_feature_view_by_version(
                    name, project, version_num, allow_cache
                )
            except NotImplementedError:
                # Fall back for v0 on registries that don't implement versioned lookup
                if version_num == 0:
                    fv = registry.get_any_feature_view(name, project, allow_cache)
                else:
                    raise
            # Set version_tag on the projection so name_to_use() returns versioned key
            if hasattr(fv, "projection") and fv.projection is not None:
                fv.projection.version_tag = version_num
        else:
            fv = registry.get_any_feature_view(name, project, allow_cache)

        if hasattr(fv, "enabled") and not fv.enabled:
            raise ValueError(
                f"Feature view '{name}' is disabled and cannot serve features. "