saving progress

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
feast-dev · franciscojavierarceo · Oct 10, 2024 · Sep 24, 2024 · Aug 17, 2024 · Sep 24, 2024
commit 47974c2cc92d79e3ce3b270d5a74d1ef99c47de5
@@ -1444,19 +1444,18 @@ def write_to_online_store(
             inputs: Optional the dictionary object to be written
             allow_registry_cache (optional): Whether to allow retrieving feature views from a cached registry.
         """
-        # TODO: restrict this to work with online StreamFeatureViews and validate the FeatureView type
+        feature_view_dict = {
+            fv_proto.name: fv_proto
+            for fv_proto in self.list_all_feature_views(allow_registry_cache)
+        }
         try:
-            feature_view: FeatureView = self.get_stream_feature_view(
-                feature_view_name, allow_registry_cache=allow_registry_cache
-            )
+            feature_view = feature_view_dict[feature_view_name]
         except FeatureViewNotFoundException:
-            feature_view = self.get_feature_view(
-                feature_view_name, allow_registry_cache=allow_registry_cache
-            )
+            raise FeatureViewNotFoundException(feature_view_name, self.project)
         if df is not None and inputs is not None:
             raise ValueError("Both df and inputs cannot be provided at the same time.")
         if df is None and inputs is not None:
-            if isinstance(inputs, dict):
+            if isinstance(inputs, dict) or isinstance(inputs, List):
                 try:
                     df = pd.DataFrame(inputs)
                 except Exception as _:
@@ -1465,6 +1464,13 @@ def write_to_online_store(
                 pass
             else:
                 raise ValueError("inputs must be a dictionary or a pandas DataFrame.")
+        if df is not None and inputs is None:
+            if isinstance(df, dict) or isinstance(df, List):
+                try:
+                    df = pd.DataFrame(df)
+                except Exception as _:
+                    raise DataFrameSerializationError
+
         provider = self._get_provider()
         provider.ingest_df(feature_view, df)
 

@@ -279,6 +279,10 @@ def ingest_df(
     ):
         table = pa.Table.from_pandas(df)
 
+        # TODO: Update this to support On Demand Feature Views.
+        # Note: A dictionary mapping of column names in this data
+        #   source to feature names in a feature table or view. Only used for feature
+        #   columns, not entity or timestamp columns.
         if feature_view.batch_source.field_mapping is not None:
             table = _run_pyarrow_field_mapping(
                 table, feature_view.batch_source.field_mapping

@@ -1,6 +1,7 @@
 import os
 import re
 import tempfile
+import time
 import unittest
 from datetime import datetime, timedelta
 from typing import Any
@@ -285,9 +286,16 @@ def test_stored_writes(self):
         entity_rows = [
             {
                 "driver_id": 1001,
+                "conv_rate": 0.25,
+                "acc_rate": 0.25,
             }
         ]
 
+        self.store.write_to_online_store(
+            feature_view_name="python_stored_writes_feature_view",
+            df=entity_rows,
+        )
+        time.sleep(1)
         online_python_response = self.store.get_online_features(
             entity_rows=entity_rows,
             features=[