Skip to content

Commit 91058c7

Browse files
msistla96msistla96
committed
refactor code, tests to include changes for other registries
1 parent f8d3c85 commit 91058c7

18 files changed

+189
-253
lines changed

‎sdk/python/feast/base_feature_view.py‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,3 @@ def with_projection(self, feature_view_projection: FeatureViewProjection):
238238
cp.projection = feature_view_projection
239239

240240
return cp
241-
242-
def update_meta(self, serialized_proto: bytes):
243-
pass

‎sdk/python/feast/entity.py‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -167,10 +167,6 @@ def from_proto(cls, entity_proto: EntityProto):
167167

168168
return entity
169169

170-
def update_meta(self, stored_proto: bytes):
171-
entity_proto = EntityProto.FromString(stored_proto)
172-
self.created_timestamp = entity_proto.meta.created_timestamp.ToDatetime()
173-
174170
def to_proto(self) -> EntityProto:
175171
"""
176172
Converts an entity object to its protobuf representation.

‎sdk/python/feast/feature_service.py‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -219,12 +219,6 @@ def from_proto(cls, feature_service_proto: FeatureServiceProto):
219219

220220
return fs
221221

222-
def update_meta(self, stored_proto: bytes):
223-
feature_service_proto = FeatureServiceProto.FromString(stored_proto)
224-
self.created_timestamp = (
225-
feature_service_proto.meta.created_timestamp.ToDatetime()
226-
)
227-
228222
def to_proto(self) -> FeatureServiceProto:
229223
"""
230224
Converts a feature service to its protobuf representation.

‎sdk/python/feast/feature_view.py‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -311,14 +311,12 @@ def with_join_key_map(self, join_key_map: Dict[str, str]):
311311

312312
return cp
313313

314-
def update_meta(self, stored_proto: bytes):
315-
feature_view_proto = FeatureViewProto.FromString(stored_proto)
316-
self.created_timestamp = feature_view_proto.meta.created_timestamp.ToDatetime()
317-
318-
for interval in feature_view_proto.meta.materialization_intervals:
319-
self.materialization_intervals.append(
320-
(interval.start_time.ToDatetime(), interval.end_time.ToDatetime())
321-
)
314+
def update_materialization_intervals(self, existing_materialization_intervals):
315+
if existing_materialization_intervals:
316+
for interval in existing_materialization_intervals:
317+
self.materialization_intervals.append(
318+
(interval.start_time.ToDatetime(), interval.end_time.ToDatetime())
319+
)
322320

323321
def to_proto(self) -> FeatureViewProto:
324322
"""

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

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,19 @@
2929
from feast.infra.infra_object import Infra
3030
from feast.on_demand_feature_view import OnDemandFeatureView
3131
from feast.project_metadata import ProjectMetadata
32+
from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto
33+
from feast.protos.feast.core.FeatureService_pb2 import (
34+
FeatureService as FeatureServiceProto,
35+
)
36+
from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
37+
from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
38+
OnDemandFeatureView as OnDemandFeatureViewProto,
39+
)
3240
from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
41+
from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto
42+
from feast.protos.feast.core.StreamFeatureView_pb2 import (
43+
StreamFeatureView as StreamFeatureViewProto,
44+
)
3345
from feast.saved_dataset import SavedDataset, ValidationReference
3446
from feast.stream_feature_view import StreamFeatureView
3547
from feast.transformation.pandas_transformation import PandasTransformation
@@ -678,3 +690,19 @@ def to_dict(self, project: str) -> Dict[str, List[Any]]:
678690
self._message_to_sorted_dict(infra_object.to_proto())
679691
)
680692
return registry_dict
693+
694+
@staticmethod
695+
def deserialize_registry_values(serialized_proto, feast_obj_type) -> Any:
696+
if feast_obj_type == Entity:
697+
return EntityProto.FromString(serialized_proto)
698+
if feast_obj_type == SavedDataset:
699+
return SavedDatasetProto.FromString(serialized_proto)
700+
if feast_obj_type == FeatureView:
701+
return FeatureViewProto.FromString(serialized_proto)
702+
if feast_obj_type == StreamFeatureView:
703+
return StreamFeatureViewProto.FromString(serialized_proto)
704+
if feast_obj_type == OnDemandFeatureView:
705+
return OnDemandFeatureViewProto.FromString(serialized_proto)
706+
if feast_obj_type == FeatureService:
707+
return FeatureServiceProto.FromString(serialized_proto)
708+
return None

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

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -265,9 +265,13 @@ def apply_entity(self, entity: Entity, project: str, commit: bool = True):
265265
existing_entity_proto.spec.name == entity_proto.spec.name
266266
and existing_entity_proto.spec.project == project
267267
):
268+
entity.created_timestamp = (
269+
existing_entity_proto.meta.created_timestamp.ToDatetime()
270+
)
271+
entity_proto = entity.to_proto()
272+
entity_proto.spec.project = project
268273
del self.cached_registry_proto.entities[idx]
269274
break
270-
271275
self.cached_registry_proto.entities.append(entity_proto)
272276
if commit:
273277
self.commit()
@@ -338,6 +342,11 @@ def apply_feature_service(
338342
== feature_service_proto.spec.name
339343
and existing_feature_service_proto.spec.project == project
340344
):
345+
feature_service.created_timestamp = (
346+
existing_feature_service_proto.meta.created_timestamp.ToDatetime()
347+
)
348+
feature_service_proto = feature_service.to_proto()
349+
feature_service_proto.spec.project = project
341350
del registry.feature_services[idx]
342351
registry.feature_services.append(feature_service_proto)
343352
if commit:
@@ -410,6 +419,18 @@ def apply_feature_view(
410419
):
411420
return
412421
else:
422+
existing_feature_view = type(feature_view).from_proto(
423+
existing_feature_view_proto
424+
)
425+
feature_view.created_timestamp = (
426+
existing_feature_view.created_timestamp.replace(tzinfo=None)
427+
)
428+
if isinstance(feature_view, (FeatureView, StreamFeatureView)):
429+
feature_view.update_materialization_intervals(
430+
existing_feature_view_proto.meta.materialization_intervals
431+
)
432+
feature_view_proto = feature_view.to_proto()
433+
feature_view_proto.spec.project = project
413434
del existing_feature_views_of_same_type[idx]
414435
break
415436

@@ -638,6 +659,17 @@ def apply_saved_dataset(
638659
existing_saved_dataset_proto.spec.name == saved_dataset_proto.spec.name
639660
and existing_saved_dataset_proto.spec.project == project
640661
):
662+
saved_dataset.created_timestamp = (
663+
existing_saved_dataset_proto.meta.created_timestamp.ToDatetime()
664+
)
665+
saved_dataset.min_event_timestamp = (
666+
existing_saved_dataset_proto.meta.min_event_timestamp.ToDatetime()
667+
)
668+
saved_dataset.max_event_timestamp = (
669+
existing_saved_dataset_proto.meta.max_event_timestamp.ToDatetime()
670+
)
671+
saved_dataset_proto = saved_dataset.to_proto()
672+
saved_dataset_proto.spec.project = project
641673
del self.cached_registry_proto.saved_datasets[idx]
642674
break
643675

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

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -693,7 +693,16 @@ def _apply_object(
693693
"feature_view_proto",
694694
"feature_service_proto",
695695
]:
696-
obj.update_meta(row._mapping[proto_field_name])
696+
deserialized_proto = self.deserialize_registry_values(
697+
row._mapping[proto_field_name], type(obj)
698+
)
699+
obj.created_timestamp = (
700+
deserialized_proto.meta.created_timestamp.ToDatetime()
701+
)
702+
if isinstance(obj, (FeatureView, StreamFeatureView)):
703+
obj.update_materialization_intervals(
704+
deserialized_proto.meta.materialization_intervals
705+
)
697706
values = {
698707
proto_field_name: obj.to_proto().SerializeToString(),
699708
"last_updated_timestamp": update_time,

‎sdk/python/feast/on_demand_feature_view.py‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -201,12 +201,6 @@ def __eq__(self, other):
201201
def __hash__(self):
202202
return super().__hash__()
203203

204-
def update_meta(self, stored_proto: bytes):
205-
on_demand_feature_view_proto = OnDemandFeatureViewProto.FromString(stored_proto)
206-
self.created_timestamp = (
207-
on_demand_feature_view_proto.meta.created_timestamp.ToDatetime()
208-
)
209-
210204
def to_proto(self) -> OnDemandFeatureViewProto:
211205
"""
212206
Converts an on demand feature view object to its protobuf representation.

‎sdk/python/feast/saved_dataset.py‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -140,16 +140,6 @@ def __eq__(self, other):
140140

141141
return True
142142

143-
def update_meta(self, stored_proto: bytes):
144-
saved_dataset_proto = SavedDatasetProto.FromString(stored_proto)
145-
self.created_timestamp = saved_dataset_proto.meta.created_timestamp.ToDatetime()
146-
self.min_event_timestamp = (
147-
saved_dataset_proto.meta.min_event_timestamp.ToDatetime()
148-
)
149-
self.max_event_timestamp = (
150-
saved_dataset_proto.meta.max_event_timestamp.ToDatetime()
151-
)
152-
153143
@staticmethod
154144
def from_proto(saved_dataset_proto: SavedDatasetProto):
155145
"""

‎sdk/python/feast/stream_feature_view.py‎

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -167,20 +167,6 @@ def __eq__(self, other):
167167
def __hash__(self) -> int:
168168
return super().__hash__()
169169

170-
def update_meta(self, stored_proto: bytes):
171-
stream_feature_view_proto = StreamFeatureViewProto.FromString(stored_proto)
172-
self.created_timestamp = (
173-
stream_feature_view_proto.meta.created_timestamp.ToDatetime()
174-
)
175-
176-
for interval in stream_feature_view_proto.meta.materialization_intervals:
177-
self.materialization_intervals.append(
178-
(
179-
interval.start_time.ToDatetime(),
180-
interval.end_time.ToDatetime(),
181-
)
182-
)
183-
184170
def to_proto(self):
185171
meta = self.to_proto_meta()
186172
ttl_duration = self.get_ttl_duration()

0 commit comments

Comments
 (0)