Skip to content

Commit bc8ef89

Browse files
ntkatholentkathole
committed
feat: Add Feature Service Create in UI
Signed-off-by: ntkathole <[email protected]>
1 parent e0a8573 commit bc8ef89

22 files changed

Lines changed: 1849 additions & 546 deletions

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

Lines changed: 86 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
1-
from typing import Dict
1+
from typing import Dict, List, Optional
22

33
from fastapi import APIRouter, Depends, Query
4+
from fastapi.responses import JSONResponse
5+
from pydantic import BaseModel
46

57
from feast.api.registry.rest.codegen_utils import render_feature_service_code
68
from feast.api.registry.rest.rest_utils import (
@@ -14,9 +16,39 @@
1416
grpc_call,
1517
parse_tags,
1618
)
19+
from feast.feature_service import FeatureService
1720
from feast.protos.feast.registry import RegistryServer_pb2
1821

1922

23+
class FeatureViewRefModel(BaseModel):
24+
feature_view_name: str
25+
feature_names: Optional[List[str]] = []
26+
27+
28+
class ApplyFeatureServiceRequestBody(BaseModel):
29+
name: str
30+
project: str
31+
features: List[FeatureViewRefModel]
32+
description: Optional[str] = ""
33+
tags: Optional[Dict[str, str]] = {}
34+
owner: Optional[str] = ""
35+
36+
37+
def _projection_view_name(projection: dict) -> Optional[str]:
38+
return (
39+
projection.get("featureViewName")
40+
or projection.get("feature_view_name")
41+
or projection.get("name")
42+
)
43+
44+
45+
def _projection_feature_columns(projection: dict) -> list:
46+
columns = projection.get("featureColumns")
47+
if columns is None:
48+
columns = projection.get("feature_columns")
49+
return columns if isinstance(columns, list) else []
50+
51+
2052
def get_feature_service_router(grpc_handler) -> APIRouter:
2153
router = APIRouter()
2254

@@ -97,9 +129,7 @@ def get_feature_service(
97129
project=project,
98130
allow_cache=allow_cache,
99131
)
100-
feature_service = grpc_call(grpc_handler.GetFeatureService, req)
101-
102-
result = feature_service
132+
result = grpc_call(grpc_handler.GetFeatureService, req)
103133

104134
if include_relationships:
105135
relationships = get_object_relationships(
@@ -109,33 +139,41 @@ def get_feature_service(
109139

110140
if result:
111141
spec = result.get("spec", result)
112-
name = spec.get("name") or result.get("name") or "default_feature_service"
142+
service_name = (
143+
spec.get("name") or result.get("name") or "default_feature_service"
144+
)
113145
projections = spec.get("features", [])
114146
if not isinstance(projections, list):
115147
projections = []
116148

117149
features_exprs = []
118150
for proj in projections:
119-
if isinstance(proj, dict):
120-
view_name = proj.get("name")
121-
feature_names = proj.get("features", [])
122-
else:
123-
view_name = str(proj)
124-
feature_names = []
151+
if not isinstance(proj, dict):
152+
continue
125153

154+
view_name = _projection_view_name(proj)
126155
if not view_name:
127156
continue
128157

158+
feature_columns = _projection_feature_columns(proj)
159+
feature_names = [
160+
column.get("name")
161+
for column in feature_columns
162+
if isinstance(column, dict) and column.get("name")
163+
]
164+
129165
if feature_names:
130-
feature_list = ", ".join([repr(f) for f in feature_names])
166+
feature_list = ", ".join(
167+
[repr(feature_name) for feature_name in feature_names]
168+
)
131169
features_exprs.append(f"{view_name}[[{feature_list}]]")
132170
else:
133171
features_exprs.append(view_name)
134172

135173
features_str = ", ".join(features_exprs)
136174

137175
context = {
138-
"name": name,
176+
"name": service_name,
139177
"features": features_str,
140178
"tags": spec.get("tags", {}),
141179
"description": spec.get("description", ""),
@@ -145,4 +183,39 @@ def get_feature_service(
145183
result["featureDefinition"] = render_feature_service_code(context)
146184
return result
147185

186+
@router.post("/feature_services", status_code=201)
187+
def apply_feature_service(body: ApplyFeatureServiceRequestBody):
188+
req = FeatureService.build_apply_request(
189+
name=body.name,
190+
project=body.project,
191+
feature_view_refs=[
192+
(feature.feature_view_name, feature.feature_names or None)
193+
for feature in body.features
194+
],
195+
description=body.description or "",
196+
tags=body.tags or {},
197+
owner=body.owner or "",
198+
commit=True,
199+
)
200+
grpc_call(grpc_handler.ApplyFeatureService, req)
201+
202+
return JSONResponse(
203+
status_code=201,
204+
content={"name": body.name, "project": body.project, "status": "applied"},
205+
)
206+
207+
@router.delete("/feature_services/{name}")
208+
def delete_feature_service(
209+
name: str,
210+
project: str = Query(...),
211+
):
212+
req = RegistryServer_pb2.DeleteFeatureServiceRequest(
213+
name=name,
214+
project=project,
215+
commit=True,
216+
)
217+
grpc_call(grpc_handler.DeleteFeatureService, req)
218+
219+
return {"name": name, "project": project, "status": "deleted"}
220+
148221
return router

‎sdk/python/feast/feature_service.py‎

Lines changed: 164 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,18 @@
11
from datetime import datetime
2-
from typing import Dict, List, Optional, Union
2+
from typing import TYPE_CHECKING, Dict, List, Optional, Union
33

44
from google.protobuf.json_format import MessageToJson
55
from typeguard import typechecked
66

77
from feast.base_feature_view import BaseFeatureView
8-
from feast.errors import FeatureViewMissingDuringFeatureServiceInference
8+
from feast.errors import (
9+
FeastObjectNotFoundException,
10+
FeatureViewMissingDuringFeatureServiceInference,
11+
)
912
from feast.feature_logging import LoggingConfig
1013
from feast.feature_view import FeatureView
1114
from feast.feature_view_projection import FeatureViewProjection
15+
from feast.field import Field
1216
from feast.labeling.label_view import LabelView
1317
from feast.on_demand_feature_view import OnDemandFeatureView
1418
from feast.protos.feast.core.FeatureService_pb2 import (
@@ -21,6 +25,9 @@
2125
FeatureServiceSpec as FeatureServiceSpecProto,
2226
)
2327

28+
if TYPE_CHECKING:
29+
from feast.infra.registry.base_registry import BaseRegistry
30+
2431

2532
@typechecked
2633
class FeatureService:
@@ -161,6 +168,119 @@ def infer_features(
161168
f'{type(feature_grouping)} as part of the "features" argument.)'
162169
)
163170

171+
def prepare_for_apply(
172+
self,
173+
registry: "BaseRegistry",
174+
project: str,
175+
allow_cache: bool = False,
176+
) -> "FeatureService":
177+
"""
178+
Materialize feature view projections before registry apply.
179+
180+
Uses the same FeatureService construction and ``infer_features`` path as
181+
``FeatureStore.apply`` for SDK-defined services.
182+
183+
When the service is already fully resolved (SDK path where _features is
184+
set and projections already have features populated via infer_features,
185+
OR the proto deserialization path where projections carry full dtype
186+
info), this is a no-op.
187+
"""
188+
from feast.types import Invalid
189+
190+
if self._features and all(p.features for p in self.feature_view_projections):
191+
return self
192+
193+
if (
194+
not self._features
195+
and self.feature_view_projections
196+
and all(
197+
p.features and all(f.dtype != Invalid for f in p.features)
198+
for p in self.feature_view_projections
199+
)
200+
):
201+
return self
202+
203+
fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]] = {}
204+
205+
if self._features:
206+
for feature_grouping in self._features:
207+
if isinstance(feature_grouping, BaseFeatureView):
208+
fvs_to_update[feature_grouping.name] = (
209+
registry.get_any_feature_view(
210+
feature_grouping.name, project, allow_cache=allow_cache
211+
)
212+
)
213+
self.infer_features(fvs_to_update=fvs_to_update)
214+
return self
215+
216+
resolved_features: List[Union[FeatureView, OnDemandFeatureView, LabelView]] = []
217+
for projection in self.feature_view_projections:
218+
try:
219+
feature_view = registry.get_any_feature_view(
220+
projection.name, project, allow_cache=allow_cache
221+
)
222+
except FeastObjectNotFoundException as exc:
223+
raise FeastObjectNotFoundException(
224+
f"Feature view '{projection.name}' not found in project '{project}'"
225+
) from exc
226+
227+
if not isinstance(
228+
feature_view, (FeatureView, OnDemandFeatureView, LabelView)
229+
):
230+
raise ValueError(
231+
f"Cannot resolve projection for feature view '{projection.name}'"
232+
)
233+
234+
fvs_to_update[feature_view.name] = feature_view
235+
features_by_name = {
236+
feature.name: feature for feature in feature_view.features
237+
}
238+
239+
if self._projection_matches_registry_features(projection, features_by_name):
240+
resolved_features.append(feature_view.with_projection(projection))
241+
elif projection.desired_features:
242+
resolved_features.append(
243+
feature_view[list(projection.desired_features)]
244+
)
245+
elif not projection.features:
246+
resolved_features.append(feature_view)
247+
else:
248+
resolved_features.append(
249+
feature_view[[feature.name for feature in projection.features]]
250+
)
251+
252+
prepared = FeatureService(
253+
name=self.name,
254+
features=resolved_features,
255+
tags=self.tags,
256+
description=self.description,
257+
owner=self.owner,
258+
logging_config=self.logging_config,
259+
precompute_online=self.precompute_online,
260+
)
261+
prepared.created_timestamp = self.created_timestamp
262+
prepared.last_updated_timestamp = self.last_updated_timestamp
263+
prepared.infer_features(fvs_to_update=fvs_to_update)
264+
265+
self._features = prepared._features
266+
self.feature_view_projections = prepared.feature_view_projections
267+
return self
268+
269+
@staticmethod
270+
def _projection_matches_registry_features(
271+
projection: FeatureViewProjection,
272+
features_by_name: Dict[str, Field],
273+
) -> bool:
274+
if not projection.features:
275+
return False
276+
277+
for feature in projection.features:
278+
if feature.name not in features_by_name:
279+
return False
280+
if feature != features_by_name[feature.name]:
281+
return False
282+
return True
283+
164284
def __repr__(self):
165285
items = (f"{k} = {v}" for k, v in self.__dict__.items())
166286
return f"<{self.__class__.__name__}({', '.join(items)})>"
@@ -259,6 +379,48 @@ def to_proto(self) -> FeatureServiceProto:
259379

260380
return FeatureServiceProto(spec=spec, meta=meta)
261381

382+
@classmethod
383+
def build_apply_request(
384+
cls,
385+
*,
386+
name: str,
387+
project: str,
388+
feature_view_refs: List[tuple[str, Optional[List[str]]]],
389+
description: str = "",
390+
tags: Optional[Dict[str, str]] = None,
391+
owner: str = "",
392+
commit: bool = True,
393+
):
394+
"""Build an unresolved ApplyFeatureServiceRequest from feature view refs."""
395+
from feast.protos.feast.core.Feature_pb2 import FeatureSpecV2
396+
from feast.protos.feast.core.FeatureViewProjection_pb2 import (
397+
FeatureViewProjection as FeatureViewProjectionProto,
398+
)
399+
from feast.protos.feast.registry import RegistryServer_pb2
400+
401+
projections = []
402+
for feature_view_name, feature_names in feature_view_refs:
403+
projection = FeatureViewProjectionProto(
404+
feature_view_name=feature_view_name,
405+
)
406+
if feature_names:
407+
for feature_name in feature_names:
408+
projection.feature_columns.append(FeatureSpecV2(name=feature_name))
409+
projections.append(projection)
410+
411+
spec = FeatureServiceSpecProto(
412+
name=name,
413+
features=projections,
414+
tags=tags or {},
415+
description=description,
416+
owner=owner,
417+
)
418+
return RegistryServer_pb2.ApplyFeatureServiceRequest(
419+
feature_service=FeatureServiceProto(spec=spec),
420+
project=project,
421+
commit=commit,
422+
)
423+
262424
def validate(self):
263425
if not self.precompute_online:
264426
return

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -453,6 +453,7 @@ def delete_data_source(self, name: str, project: str, commit: bool = True):
453453
def apply_feature_service(
454454
self, feature_service: FeatureService, project: str, commit: bool = True
455455
):
456+
feature_service.prepare_for_apply(self, project, allow_cache=True)
456457
now = _utc_now()
457458
if not feature_service.created_timestamp:
458459
feature_service.created_timestamp = now

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -264,6 +264,7 @@ def apply_entity(self, entity: Entity, project: str, commit: bool = True):
264264
def apply_feature_service(
265265
self, feature_service: FeatureService, project: str, commit: bool = True
266266
):
267+
feature_service.prepare_for_apply(self, project, allow_cache=True)
267268
return self._apply_object(
268269
"FEATURE_SERVICES",
269270
project,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1031,6 +1031,7 @@ def apply_feature_view(
10311031
def apply_feature_service(
10321032
self, feature_service: FeatureService, project: str, commit: bool = True
10331033
):
1034+
feature_service.prepare_for_apply(self, project, allow_cache=True)
10341035
return self._apply_object(
10351036
feature_services,
10361037
project,

0 commit comments

Comments
 (0)