feat: Add /write-to-online-store method to the python feature server (#2423)

Signed-off-by: Tomas Pereira de Vasconcelos <tomasvasconcelos1@gmail.com>

Author: tpvasconcelos

sdk/python/feast/feature_server.py

@@ -1,16 +1,26 @@
+import json
 import traceback
+import warnings
 
+import pandas as pd
 import uvicorn
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.logger import logger
 from fastapi.params import Depends
 from google.protobuf.json_format import MessageToDict, Parse
+from pydantic import BaseModel
 
 import feast
 from feast import proto_json
 from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest
 
 
+class WriteToFeatureStoreRequest(BaseModel):
+    feature_view_name: str
+    df: dict
+    allow_registry_cache: bool = True
+
+
 def get_app(store: "feast.FeatureStore"):
     proto_json.patch()
 
@@ -58,6 +68,28 @@ def get_online_features(body=Depends(get_body)):
             # Raise HTTPException to return the error message to the client
             raise HTTPException(status_code=500, detail=str(e))
 
+    @app.post("/write-to-online-store")
+    def write_to_online_store(body=Depends(get_body)):
+        warnings.warn(
+            "write_to_online_store is an experimental feature. "
+            "This API is unstable and it could be changed in the future. "
+            "We do not guarantee that future changes will maintain backward compatibility.",
+            RuntimeWarning,
+        )
+        try:
+            request = WriteToFeatureStoreRequest(**json.loads(body))
+            df = pd.DataFrame(request.df)
+            store.write_to_online_store(
+                feature_view_name=request.feature_view_name,
+                df=df,
+                allow_registry_cache=request.allow_registry_cache,
+            )
+        except Exception as e:
+            # Print the original exception on the server side
+            logger.exception(traceback.format_exc())
+            # Raise HTTPException to return the error message to the client
+            raise HTTPException(status_code=500, detail=str(e))
+
     return app