feat: Add Feature Service Create/delete REST apis and UI

sdk/python/feast/api/registry/rest/feature_services.py

@@ -1,6 +1,8 @@
-from typing import Dict
+from typing import Dict, List, Optional
 
 from fastapi import APIRouter, Depends, Query
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
 
 from feast.api.registry.rest.codegen_utils import render_feature_service_code
 from feast.api.registry.rest.rest_utils import (
@@ -14,9 +16,39 @@
     grpc_call,
     parse_tags,
 )
+from feast.feature_service import FeatureService
 from feast.protos.feast.registry import RegistryServer_pb2
 
 
+class FeatureViewRefModel(BaseModel):
+    feature_view_name: str
+    feature_names: Optional[List[str]] = []
+
+
+class ApplyFeatureServiceRequestBody(BaseModel):
+    name: str
+    project: str
+    features: List[FeatureViewRefModel]
+    description: Optional[str] = ""
+    tags: Optional[Dict[str, str]] = {}
+    owner: Optional[str] = ""
+
+
+def _projection_view_name(projection: dict) -> Optional[str]:
+    return (
+        projection.get("featureViewName")
+        or projection.get("feature_view_name")
+        or projection.get("name")
+    )
+
+
+def _projection_feature_columns(projection: dict) -> list:
+    columns = projection.get("featureColumns")
+    if columns is None:
+        columns = projection.get("feature_columns")
+    return columns if isinstance(columns, list) else []
+
+
 def get_feature_service_router(grpc_handler) -> APIRouter:
     router = APIRouter()
 
@@ -97,9 +129,7 @@ def get_feature_service(
             project=project,
             allow_cache=allow_cache,
         )
-        feature_service = grpc_call(grpc_handler.GetFeatureService, req)
-
-        result = feature_service
+        result = grpc_call(grpc_handler.GetFeatureService, req)
 
         if include_relationships:
             relationships = get_object_relationships(
@@ -109,33 +139,41 @@ def get_feature_service(
 
         if result:
             spec = result.get("spec", result)
-            name = spec.get("name") or result.get("name") or "default_feature_service"
+            service_name = (
+                spec.get("name") or result.get("name") or "default_feature_service"
+            )
             projections = spec.get("features", [])
             if not isinstance(projections, list):
                 projections = []
 
             features_exprs = []
             for proj in projections:
-                if isinstance(proj, dict):
-                    view_name = proj.get("name")
-                    feature_names = proj.get("features", [])
-                else:
-                    view_name = str(proj)
-                    feature_names = []
+                if not isinstance(proj, dict):
+                    continue
 
+                view_name = _projection_view_name(proj)
                 if not view_name:
                     continue
 
+                feature_columns = _projection_feature_columns(proj)
+                feature_names = [
+                    column.get("name")
+                    for column in feature_columns
+                    if isinstance(column, dict) and column.get("name")
+                ]
+
                 if feature_names:
-                    feature_list = ", ".join([repr(f) for f in feature_names])
+                    feature_list = ", ".join(
+                        [repr(feature_name) for feature_name in feature_names]
+                    )
                     features_exprs.append(f"{view_name}[[{feature_list}]]")
                 else:
                     features_exprs.append(view_name)
 
             features_str = ", ".join(features_exprs)
 
             context = {
-                "name": name,
+                "name": service_name,
                 "features": features_str,
                 "tags": spec.get("tags", {}),
                 "description": spec.get("description", ""),
@@ -145,4 +183,39 @@ def get_feature_service(
             result["featureDefinition"] = render_feature_service_code(context)
         return result
 
+    @router.post("/feature_services", status_code=201)
+    def apply_feature_service(body: ApplyFeatureServiceRequestBody):
+        req = FeatureService.build_apply_request(
+            name=body.name,
+            project=body.project,
+            feature_view_refs=[
+                (feature.feature_view_name, feature.feature_names or None)
+                for feature in body.features
+            ],
+            description=body.description or "",
+            tags=body.tags or {},
+            owner=body.owner or "",
+            commit=True,
+        )
+        grpc_call(grpc_handler.ApplyFeatureService, req)
+
+        return JSONResponse(
+            status_code=201,
+            content={"name": body.name, "project": body.project, "status": "applied"},
+        )
+
+    @router.delete("/feature_services/{name}")
+    def delete_feature_service(
+        name: str,
+        project: str = Query(...),
+    ):
+        req = RegistryServer_pb2.DeleteFeatureServiceRequest(
+            name=name,
+            project=project,
+            commit=True,
+        )
+        grpc_call(grpc_handler.DeleteFeatureService, req)
+
+        return {"name": name, "project": project, "status": "deleted"}
+
     return router

sdk/python/feast/feature_service.py

@@ -1,14 +1,18 @@
 from datetime import datetime
-from typing import Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
 
 from google.protobuf.json_format import MessageToJson
 from typeguard import typechecked
 
 from feast.base_feature_view import BaseFeatureView
-from feast.errors import FeatureViewMissingDuringFeatureServiceInference
+from feast.errors import (
+    FeastObjectNotFoundException,
+    FeatureViewMissingDuringFeatureServiceInference,
+)
 from feast.feature_logging import LoggingConfig
 from feast.feature_view import FeatureView
 from feast.feature_view_projection import FeatureViewProjection
+from feast.field import Field
 from feast.labeling.label_view import LabelView
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.protos.feast.core.FeatureService_pb2 import (
@@ -21,6 +25,9 @@
     FeatureServiceSpec as FeatureServiceSpecProto,
 )
 
+if TYPE_CHECKING:
+    from feast.infra.registry.base_registry import BaseRegistry
+
 
 @typechecked
 class FeatureService:
@@ -161,6 +168,119 @@ def infer_features(
                     f'{type(feature_grouping)} as part of the "features" argument.)'
                 )
 
+    def prepare_for_apply(
+        self,
+        registry: "BaseRegistry",
+        project: str,
+        allow_cache: bool = False,
+    ) -> "FeatureService":
+        """
+        Materialize feature view projections before registry apply.
+
+        Uses the same FeatureService construction and ``infer_features`` path as
+        ``FeatureStore.apply`` for SDK-defined services.
+
+        When the service is already fully resolved (SDK path where _features is
+        set and projections already have features populated via infer_features,
+        OR the proto deserialization path where projections carry full dtype
+        info), this is a no-op.
+        """
+        from feast.types import Invalid
+
+        if self._features and all(p.features for p in self.feature_view_projections):
+            return self
+
+        if (
+            not self._features
+            and self.feature_view_projections
+            and all(
+                p.features and all(f.dtype != Invalid for f in p.features)
+                for p in self.feature_view_projections
+            )
+        ):
+            return self
+
+        fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]] = {}
+
+        if self._features:
+            for feature_grouping in self._features:
+                if isinstance(feature_grouping, BaseFeatureView):
+                    fvs_to_update[feature_grouping.name] = (
+                        registry.get_any_feature_view(
+                            feature_grouping.name, project, allow_cache=allow_cache
+                        )
+                    )
+            self.infer_features(fvs_to_update=fvs_to_update)
+            return self
+
+        resolved_features: List[Union[FeatureView, OnDemandFeatureView, LabelView]] = []
+        for projection in self.feature_view_projections:
+            try:
+                feature_view = registry.get_any_feature_view(
+                    projection.name, project, allow_cache=allow_cache
+                )
+            except FeastObjectNotFoundException as exc:
+                raise FeastObjectNotFoundException(
+                    f"Feature view '{projection.name}' not found in project '{project}'"
+                ) from exc
+
+            if not isinstance(
+                feature_view, (FeatureView, OnDemandFeatureView, LabelView)
+            ):
+                raise ValueError(
+                    f"Cannot resolve projection for feature view '{projection.name}'"
+                )
+
+            fvs_to_update[feature_view.name] = feature_view
+            features_by_name = {
+                feature.name: feature for feature in feature_view.features
+            }
+
+            if self._projection_matches_registry_features(projection, features_by_name):
+                resolved_features.append(feature_view.with_projection(projection))
+            elif projection.desired_features:
+                resolved_features.append(
+                    feature_view[list(projection.desired_features)]
+                )
+            elif not projection.features:
+                resolved_features.append(feature_view)
+            else:
+                resolved_features.append(
+                    feature_view[[feature.name for feature in projection.features]]
+                )
+
+        prepared = FeatureService(
+            name=self.name,
+            features=resolved_features,
+            tags=self.tags,
+            description=self.description,
+            owner=self.owner,
+            logging_config=self.logging_config,
+            precompute_online=self.precompute_online,
+        )
+        prepared.created_timestamp = self.created_timestamp
+        prepared.last_updated_timestamp = self.last_updated_timestamp
+        prepared.infer_features(fvs_to_update=fvs_to_update)
+
+        self._features = prepared._features
+        self.feature_view_projections = prepared.feature_view_projections
+        return self
+
+    @staticmethod
+    def _projection_matches_registry_features(
+        projection: FeatureViewProjection,
+        features_by_name: Dict[str, Field],
+    ) -> bool:
+        if not projection.features:
+            return False
+
+        for feature in projection.features:
+            if feature.name not in features_by_name:
+                return False
+            if feature != features_by_name[feature.name]:
+                return False
+        return True
+
     def __repr__(self):
         items = (f"{k} = {v}" for k, v in self.__dict__.items())
         return f"<{self.__class__.__name__}({', '.join(items)})>"
@@ -259,6 +379,48 @@ def to_proto(self) -> FeatureServiceProto:
 
         return FeatureServiceProto(spec=spec, meta=meta)
 
+    @classmethod
+    def build_apply_request(
+        cls,
+        *,
+        name: str,
+        project: str,
+        feature_view_refs: List[tuple[str, Optional[List[str]]]],
+        description: str = "",
+        tags: Optional[Dict[str, str]] = None,
+        owner: str = "",
+        commit: bool = True,
+    ):
+        """Build an unresolved ApplyFeatureServiceRequest from feature view refs."""
+        from feast.protos.feast.core.Feature_pb2 import FeatureSpecV2
+        from feast.protos.feast.core.FeatureViewProjection_pb2 import (
+            FeatureViewProjection as FeatureViewProjectionProto,
+        )
+        from feast.protos.feast.registry import RegistryServer_pb2
+
+        projections = []
+        for feature_view_name, feature_names in feature_view_refs:
+            projection = FeatureViewProjectionProto(
+                feature_view_name=feature_view_name,
+            )
+            if feature_names:
+                for feature_name in feature_names:
+                    projection.feature_columns.append(FeatureSpecV2(name=feature_name))
+            projections.append(projection)
+
+        spec = FeatureServiceSpecProto(
+            name=name,
+            features=projections,
+            tags=tags or {},
+            description=description,
+            owner=owner,
+        )
+        return RegistryServer_pb2.ApplyFeatureServiceRequest(
+            feature_service=FeatureServiceProto(spec=spec),
+            project=project,
+            commit=commit,
+        )
+
     def validate(self):
         if not self.precompute_online:
             return

sdk/python/feast/infra/registry/registry.py

@@ -453,6 +453,7 @@ def delete_data_source(self, name: str, project: str, commit: bool = True):
     def apply_feature_service(
         self, feature_service: FeatureService, project: str, commit: bool = True
     ):
+        feature_service.prepare_for_apply(self, project, allow_cache=True)
         now = _utc_now()
         if not feature_service.created_timestamp:
             feature_service.created_timestamp = now

sdk/python/feast/infra/registry/snowflake.py

@@ -264,6 +264,7 @@ def apply_entity(self, entity: Entity, project: str, commit: bool = True):
     def apply_feature_service(
         self, feature_service: FeatureService, project: str, commit: bool = True
     ):
+        feature_service.prepare_for_apply(self, project, allow_cache=True)
         return self._apply_object(
             "FEATURE_SERVICES",
             project,

sdk/python/feast/infra/registry/sql.py

@@ -1031,6 +1031,7 @@ def apply_feature_view(
     def apply_feature_service(
         self, feature_service: FeatureService, project: str, commit: bool = True
     ):
+        feature_service.prepare_for_apply(self, project, allow_cache=True)
         return self._apply_object(
             feature_services,
             project,