We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
feast serve convert number to INT64 
## Expected Behavior 
when we use the get-online-features endpoint the request body numbers converted to Int64 type and while from python file calling get_online_feature function convert number to Int32 which leads to inconsistency between both use 
if the entity value type is Int32 then feature servering with online feast serve command will not work expected behavior is to convert the value to Int32 while it is being converted to Int64
## Current Behavior
when using feast serve number are converted to Int64 type 
## Steps to reproduce
create a entity with a column with type Int32 create a feature view with the entity apply the changes using feast apply materialize the data to online store start feature server using feast serve and call endpoint /get-online-feature to retrive feature ... expected response is the feature retrived  while it show none due to type issue 
### Specifications

- Version: 0.36
- Platform: 
- Subsystem:

## Possible Solution
before converting the type to Int64 get the repo and convert the type to the entity data type defined 
other solution is just write in doc to use Int64 and Float64 type while defining entity


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/feature_server.py`
Content:
```
1 import json
2 import sys
3 import threading
4 import traceback
5 import warnings
6 from typing import List, Optional
7 
8 import pandas as pd
9 from dateutil import parser
10 from fastapi import FastAPI, HTTPException, Request, Response, status
11 from fastapi.logger import logger
12 from fastapi.params import Depends
13 from google.protobuf.json_format import MessageToDict, Parse
14 from pydantic import BaseModel
15 
16 import feast
17 from feast import proto_json, utils
18 from feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL
19 from feast.data_source import PushMode
20 from feast.errors import PushSourceNotFoundException
21 from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest
22 
23 
24 # TODO: deprecate this in favor of push features
25 class WriteToFeatureStoreRequest(BaseModel):
26     feature_view_name: str
27     df: dict
28     allow_registry_cache: bool = True
29 
30 
31 class PushFeaturesRequest(BaseModel):
32     push_source_name: str
33     df: dict
34     allow_registry_cache: bool = True
35     to: str = "online"
36 
37 
38 class MaterializeRequest(BaseModel):
39     start_ts: str
40     end_ts: str
41     feature_views: Optional[List[str]] = None
42 
43 
44 class MaterializeIncrementalRequest(BaseModel):
45     end_ts: str
46     feature_views: Optional[List[str]] = None
47 
48 
49 def get_app(
50     store: "feast.FeatureStore",
51     registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,
52 ):
53     proto_json.patch()
54 
55     app = FastAPI()
56     # Asynchronously refresh registry, notifying shutdown and canceling the active timer if the app is shutting down
57     registry_proto = None
58     shutting_down = False
59     active_timer: Optional[threading.Timer] = None
60 
61     async def get_body(request: Request):
62         return await request.body()
63 
64     def async_refresh():
65         store.refresh_registry()
66         nonlocal registry_proto
67         registry_proto = store.registry.proto()
68         if shutting_down:
69             return
70         nonlocal active_timer
71         active_timer = threading.Timer(registry_ttl_sec, async_refresh)
72         active_timer.start()
73 
74     @app.on_event("shutdown")
75     def shutdown_event():
76         nonlocal shutting_down
77         shutting_down = True
78         if active_timer:
79             active_timer.cancel()
80 
81     async_refresh()
82 
83     @app.post("/get-online-features")
84     def get_online_features(body=Depends(get_body)):
85         try:
86             # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object
87             request_proto = GetOnlineFeaturesRequest()
88             Parse(body, request_proto)
89 
90             # Initialize parameters for FeatureStore.get_online_features(...) call
91             if request_proto.HasField("feature_service"):
92                 features = store.get_feature_service(
93                     request_proto.feature_service, allow_cache=True
94                 )
95             else:
96                 features = list(request_proto.features.val)
97 
98             full_feature_names = request_proto.full_feature_names
99 
100             batch_sizes = [len(v.val) for v in request_proto.entities.values()]
101             num_entities = batch_sizes[0]
102             if any(batch_size != num_entities for batch_size in batch_sizes):
103                 raise HTTPException(status_code=500, detail="Uneven number of columns")
104 
105             response_proto = store._get_online_features(
106                 features=features,
107                 entity_values=request_proto.entities,
108                 full_feature_names=full_feature_names,
109                 native_entity_values=False,
110             ).proto
111 
112             # Convert the Protobuf object to JSON and return it
113             return MessageToDict(  # type: ignore
114                 response_proto, preserving_proto_field_name=True, float_precision=18
115             )
116         except Exception as e:
117             # Print the original exception on the server side
118             logger.exception(traceback.format_exc())
119             # Raise HTTPException to return the error message to the client
120             raise HTTPException(status_code=500, detail=str(e))
121 
122     @app.post("/push")
123     def push(body=Depends(get_body)):
124         try:
125             request = PushFeaturesRequest(**json.loads(body))
126             df = pd.DataFrame(request.df)
127             if request.to == "offline":
128                 to = PushMode.OFFLINE
129             elif request.to == "online":
130                 to = PushMode.ONLINE
131             elif request.to == "online_and_offline":
132                 to = PushMode.ONLINE_AND_OFFLINE
133             else:
134                 raise ValueError(
135                     f"{request.to} is not a supported push format. Please specify one of these ['online', 'offline', 'online_and_offline']."
136                 )
137             store.push(
138                 push_source_name=request.push_source_name,
139                 df=df,
140                 allow_registry_cache=request.allow_registry_cache,
141                 to=to,
142             )
143         except PushSourceNotFoundException as e:
144             # Print the original exception on the server side
145             logger.exception(traceback.format_exc())
146             # Raise HTTPException to return the error message to the client
147             raise HTTPException(status_code=422, detail=str(e))
148         except Exception as e:
149             # Print the original exception on the server side
150             logger.exception(traceback.format_exc())
151             # Raise HTTPException to return the error message to the client
152             raise HTTPException(status_code=500, detail=str(e))
153 
154     @app.post("/write-to-online-store")
155     def write_to_online_store(body=Depends(get_body)):
156         warnings.warn(
157             "write_to_online_store is deprecated. Please consider using /push instead",
158             RuntimeWarning,
159         )
160         try:
161             request = WriteToFeatureStoreRequest(**json.loads(body))
162             df = pd.DataFrame(request.df)
163             store.write_to_online_store(
164                 feature_view_name=request.feature_view_name,
165                 df=df,
166                 allow_registry_cache=request.allow_registry_cache,
167             )
168         except Exception as e:
169             # Print the original exception on the server side
170             logger.exception(traceback.format_exc())
171             # Raise HTTPException to return the error message to the client
172             raise HTTPException(status_code=500, detail=str(e))
173 
174     @app.get("/health")
175     def health():
176         return Response(status_code=status.HTTP_200_OK)
177 
178     @app.post("/materialize")
179     def materialize(body=Depends(get_body)):
180         try:
181             request = MaterializeRequest(**json.loads(body))
182             store.materialize(
183                 utils.make_tzaware(parser.parse(request.start_ts)),
184                 utils.make_tzaware(parser.parse(request.end_ts)),
185                 request.feature_views,
186             )
187         except Exception as e:
188             # Print the original exception on the server side
189             logger.exception(traceback.format_exc())
190             # Raise HTTPException to return the error message to the client
191             raise HTTPException(status_code=500, detail=str(e))
192 
193     @app.post("/materialize-incremental")
194     def materialize_incremental(body=Depends(get_body)):
195         try:
196             request = MaterializeIncrementalRequest(**json.loads(body))
197             store.materialize_incremental(
198                 utils.make_tzaware(parser.parse(request.end_ts)), request.feature_views
199             )
200         except Exception as e:
201             # Print the original exception on the server side
202             logger.exception(traceback.format_exc())
203             # Raise HTTPException to return the error message to the client
204             raise HTTPException(status_code=500, detail=str(e))
205 
206     return app
207 
208 
209 if sys.platform != "win32":
210     import gunicorn.app.base
211 
212     class FeastServeApplication(gunicorn.app.base.BaseApplication):
213         def __init__(self, store: "feast.FeatureStore", **options):
214             self._app = get_app(
215                 store=store,
216                 registry_ttl_sec=options["registry_ttl_sec"],
217             )
218             self._options = options
219             super().__init__()
220 
221         def load_config(self):
222             for key, value in self._options.items():
223                 if key.lower() in self.cfg.settings and value is not None:
224                     self.cfg.set(key.lower(), value)
225 
226             self.cfg.set("worker_class", "uvicorn.workers.UvicornWorker")
227 
228         def load(self):
229             return self._app
230 
231 
232 def start_server(
233     store: "feast.FeatureStore",
234     host: str,
235     port: int,
236     no_access_log: bool,
237     workers: int,
238     keep_alive_timeout: int,
239     registry_ttl_sec: int,
240 ):
241     if sys.platform != "win32":
242         FeastServeApplication(
243             store=store,
244             bind=f"{host}:{port}",
245             accesslog=None if no_access_log else "-",
246             workers=workers,
247             keepalive=keep_alive_timeout,
248             registry_ttl_sec=registry_ttl_sec,
249         ).run()
250     else:
251         import uvicorn
252 
253         app = get_app(store, registry_ttl_sec)
254         uvicorn.run(app, host=host, port=port, access_log=(not no_access_log))
255
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py
--- a/sdk/python/feast/feature_server.py
+++ b/sdk/python/feast/feature_server.py
@@ -10,7 +10,7 @@
 from fastapi import FastAPI, HTTPException, Request, Response, status
 from fastapi.logger import logger
 from fastapi.params import Depends
-from google.protobuf.json_format import MessageToDict, Parse
+from google.protobuf.json_format import MessageToDict
 from pydantic import BaseModel
 
 import feast
@@ -18,7 +18,6 @@
 from feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL
 from feast.data_source import PushMode
 from feast.errors import PushSourceNotFoundException
-from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest
 
 
 # TODO: deprecate this in favor of push features
@@ -83,34 +82,25 @@
     @app.post("/get-online-features")
     def get_online_features(body=Depends(get_body)):
         try:
-            # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object
-            request_proto = GetOnlineFeaturesRequest()
-            Parse(body, request_proto)
-
+            body = json.loads(body)
             # Initialize parameters for FeatureStore.get_online_features(...) call
-            if request_proto.HasField("feature_service"):
+            if "feature_service" in body:
                 features = store.get_feature_service(
-                    request_proto.feature_service, allow_cache=True
+                    body["feature_service"], allow_cache=True
                 )
             else:
-                features = list(request_proto.features.val)
-
-            full_feature_names = request_proto.full_feature_names
+                features = body["features"]
 
-            batch_sizes = [len(v.val) for v in request_proto.entities.values()]
-            num_entities = batch_sizes[0]
-            if any(batch_size != num_entities for batch_size in batch_sizes):
-                raise HTTPException(status_code=500, detail="Uneven number of columns")
+            full_feature_names = body.get("full_feature_names", False)
 
             response_proto = store._get_online_features(
                 features=features,
-                entity_values=request_proto.entities,
+                entity_values=body["entities"],
                 full_feature_names=full_feature_names,
-                native_entity_values=False,
             ).proto
 
             # Convert the Protobuf object to JSON and return it
-            return MessageToDict(  # type: ignore
+            return MessageToDict(
                 response_proto, preserving_proto_field_name=True, float_precision=18
             )
         except Exception as e:

{"golden_diff": "diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py\n--- a/sdk/python/feast/feature_server.py\n+++ b/sdk/python/feast/feature_server.py\n@@ -10,7 +10,7 @@\n from fastapi import FastAPI, HTTPException, Request, Response, status\n from fastapi.logger import logger\n from fastapi.params import Depends\n-from google.protobuf.json_format import MessageToDict, Parse\n+from google.protobuf.json_format import MessageToDict\n from pydantic import BaseModel\n \n import feast\n@@ -18,7 +18,6 @@\n from feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL\n from feast.data_source import PushMode\n from feast.errors import PushSourceNotFoundException\n-from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest\n \n \n # TODO: deprecate this in favor of push features\n@@ -83,34 +82,25 @@\n     @app.post(\"/get-online-features\")\n     def get_online_features(body=Depends(get_body)):\n         try:\n-            # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object\n-            request_proto = GetOnlineFeaturesRequest()\n-            Parse(body, request_proto)\n-\n+            body = json.loads(body)\n             # Initialize parameters for FeatureStore.get_online_features(...) call\n-            if request_proto.HasField(\"feature_service\"):\n+            if \"feature_service\" in body:\n                 features = store.get_feature_service(\n-                    request_proto.feature_service, allow_cache=True\n+                    body[\"feature_service\"], allow_cache=True\n                 )\n             else:\n-                features = list(request_proto.features.val)\n-\n-            full_feature_names = request_proto.full_feature_names\n+                features = body[\"features\"]\n \n-            batch_sizes = [len(v.val) for v in request_proto.entities.values()]\n-            num_entities = batch_sizes[0]\n-            if any(batch_size != num_entities for batch_size in batch_sizes):\n-                raise HTTPException(status_code=500, detail=\"Uneven number of columns\")\n+            full_feature_names = body.get(\"full_feature_names\", False)\n \n             response_proto = store._get_online_features(\n                 features=features,\n-                entity_values=request_proto.entities,\n+                entity_values=body[\"entities\"],\n                 full_feature_names=full_feature_names,\n-                native_entity_values=False,\n             ).proto\n \n             # Convert the Protobuf object to JSON and return it\n-            return MessageToDict(  # type: ignore\n+            return MessageToDict(\n                 response_proto, preserving_proto_field_name=True, float_precision=18\n             )\n         except Exception as e:\n", "issue": "feast serve convert number to INT64 \n## Expected Behavior \r\nwhen we use the get-online-features endpoint the request body numbers converted to Int64 type and while from python file calling get_online_feature function convert number to Int32 which leads to inconsistency between both use \r\nif the entity value type is Int32 then feature servering with online feast serve command will not work expected behavior is to convert the value to Int32 while it is being converted to Int64\r\n## Current Behavior\r\nwhen using feast serve number are converted to Int64 type \r\n## Steps to reproduce\r\ncreate a entity with a column with type Int32 create a feature view with the entity apply the changes using feast apply materialize the data to online store start feature server using feast serve and call endpoint /get-online-feature to retrive feature ... expected response is the feature retrived  while it show none due to type issue \r\n### Specifications\r\n\r\n- Version: 0.36\r\n- Platform: \r\n- Subsystem:\r\n\r\n## Possible Solution\r\nbefore converting the type to Int64 get the repo and convert the type to the entity data type defined \r\nother solution is just write in doc to use Int64 and Float64 type while defining entity\r\n\n", "before_files": [{"content": "import json\nimport sys\nimport threading\nimport traceback\nimport warnings\nfrom typing import List, Optional\n\nimport pandas as pd\nfrom dateutil import parser\nfrom fastapi import FastAPI, HTTPException, Request, Response, status\nfrom fastapi.logger import logger\nfrom fastapi.params import Depends\nfrom google.protobuf.json_format import MessageToDict, Parse\nfrom pydantic import BaseModel\n\nimport feast\nfrom feast import proto_json, utils\nfrom feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL\nfrom feast.data_source import PushMode\nfrom feast.errors import PushSourceNotFoundException\nfrom feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest\n\n\n# TODO: deprecate this in favor of push features\nclass WriteToFeatureStoreRequest(BaseModel):\n    feature_view_name: str\n    df: dict\n    allow_registry_cache: bool = True\n\n\nclass PushFeaturesRequest(BaseModel):\n    push_source_name: str\n    df: dict\n    allow_registry_cache: bool = True\n    to: str = \"online\"\n\n\nclass MaterializeRequest(BaseModel):\n    start_ts: str\n    end_ts: str\n    feature_views: Optional[List[str]] = None\n\n\nclass MaterializeIncrementalRequest(BaseModel):\n    end_ts: str\n    feature_views: Optional[List[str]] = None\n\n\ndef get_app(\n    store: \"feast.FeatureStore\",\n    registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,\n):\n    proto_json.patch()\n\n    app = FastAPI()\n    # Asynchronously refresh registry, notifying shutdown and canceling the active timer if the app is shutting down\n    registry_proto = None\n    shutting_down = False\n    active_timer: Optional[threading.Timer] = None\n\n    async def get_body(request: Request):\n        return await request.body()\n\n    def async_refresh():\n        store.refresh_registry()\n        nonlocal registry_proto\n        registry_proto = store.registry.proto()\n        if shutting_down:\n            return\n        nonlocal active_timer\n        active_timer = threading.Timer(registry_ttl_sec, async_refresh)\n        active_timer.start()\n\n    @app.on_event(\"shutdown\")\n    def shutdown_event():\n        nonlocal shutting_down\n        shutting_down = True\n        if active_timer:\n            active_timer.cancel()\n\n    async_refresh()\n\n    @app.post(\"/get-online-features\")\n    def get_online_features(body=Depends(get_body)):\n        try:\n            # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object\n            request_proto = GetOnlineFeaturesRequest()\n            Parse(body, request_proto)\n\n            # Initialize parameters for FeatureStore.get_online_features(...) call\n            if request_proto.HasField(\"feature_service\"):\n                features = store.get_feature_service(\n                    request_proto.feature_service, allow_cache=True\n                )\n            else:\n                features = list(request_proto.features.val)\n\n            full_feature_names = request_proto.full_feature_names\n\n            batch_sizes = [len(v.val) for v in request_proto.entities.values()]\n            num_entities = batch_sizes[0]\n            if any(batch_size != num_entities for batch_size in batch_sizes):\n                raise HTTPException(status_code=500, detail=\"Uneven number of columns\")\n\n            response_proto = store._get_online_features(\n                features=features,\n                entity_values=request_proto.entities,\n                full_feature_names=full_feature_names,\n                native_entity_values=False,\n            ).proto\n\n            # Convert the Protobuf object to JSON and return it\n            return MessageToDict(  # type: ignore\n                response_proto, preserving_proto_field_name=True, float_precision=18\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/push\")\n    def push(body=Depends(get_body)):\n        try:\n            request = PushFeaturesRequest(**json.loads(body))\n            df = pd.DataFrame(request.df)\n            if request.to == \"offline\":\n                to = PushMode.OFFLINE\n            elif request.to == \"online\":\n                to = PushMode.ONLINE\n            elif request.to == \"online_and_offline\":\n                to = PushMode.ONLINE_AND_OFFLINE\n            else:\n                raise ValueError(\n                    f\"{request.to} is not a supported push format. Please specify one of these ['online', 'offline', 'online_and_offline'].\"\n                )\n            store.push(\n                push_source_name=request.push_source_name,\n                df=df,\n                allow_registry_cache=request.allow_registry_cache,\n                to=to,\n            )\n        except PushSourceNotFoundException as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=422, detail=str(e))\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/write-to-online-store\")\n    def write_to_online_store(body=Depends(get_body)):\n        warnings.warn(\n            \"write_to_online_store is deprecated. Please consider using /push instead\",\n            RuntimeWarning,\n        )\n        try:\n            request = WriteToFeatureStoreRequest(**json.loads(body))\n            df = pd.DataFrame(request.df)\n            store.write_to_online_store(\n                feature_view_name=request.feature_view_name,\n                df=df,\n                allow_registry_cache=request.allow_registry_cache,\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.get(\"/health\")\n    def health():\n        return Response(status_code=status.HTTP_200_OK)\n\n    @app.post(\"/materialize\")\n    def materialize(body=Depends(get_body)):\n        try:\n            request = MaterializeRequest(**json.loads(body))\n            store.materialize(\n                utils.make_tzaware(parser.parse(request.start_ts)),\n                utils.make_tzaware(parser.parse(request.end_ts)),\n                request.feature_views,\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/materialize-incremental\")\n    def materialize_incremental(body=Depends(get_body)):\n        try:\n            request = MaterializeIncrementalRequest(**json.loads(body))\n            store.materialize_incremental(\n                utils.make_tzaware(parser.parse(request.end_ts)), request.feature_views\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    return app\n\n\nif sys.platform != \"win32\":\n    import gunicorn.app.base\n\n    class FeastServeApplication(gunicorn.app.base.BaseApplication):\n        def __init__(self, store: \"feast.FeatureStore\", **options):\n            self._app = get_app(\n                store=store,\n                registry_ttl_sec=options[\"registry_ttl_sec\"],\n            )\n            self._options = options\n            super().__init__()\n\n        def load_config(self):\n            for key, value in self._options.items():\n                if key.lower() in self.cfg.settings and value is not None:\n                    self.cfg.set(key.lower(), value)\n\n            self.cfg.set(\"worker_class\", \"uvicorn.workers.UvicornWorker\")\n\n        def load(self):\n            return self._app\n\n\ndef start_server(\n    store: \"feast.FeatureStore\",\n    host: str,\n    port: int,\n    no_access_log: bool,\n    workers: int,\n    keep_alive_timeout: int,\n    registry_ttl_sec: int,\n):\n    if sys.platform != \"win32\":\n        FeastServeApplication(\n            store=store,\n            bind=f\"{host}:{port}\",\n            accesslog=None if no_access_log else \"-\",\n            workers=workers,\n            keepalive=keep_alive_timeout,\n            registry_ttl_sec=registry_ttl_sec,\n        ).run()\n    else:\n        import uvicorn\n\n        app = get_app(store, registry_ttl_sec)\n        uvicorn.run(app, host=host, port=port, access_log=(not no_access_log))\n", "path": "sdk/python/feast/feature_server.py"}], "after_files": [{"content": "import json\nimport sys\nimport threading\nimport traceback\nimport warnings\nfrom typing import List, Optional\n\nimport pandas as pd\nfrom dateutil import parser\nfrom fastapi import FastAPI, HTTPException, Request, Response, status\nfrom fastapi.logger import logger\nfrom fastapi.params import Depends\nfrom google.protobuf.json_format import MessageToDict\nfrom pydantic import BaseModel\n\nimport feast\nfrom feast import proto_json, utils\nfrom feast.constants import DEFAULT_FEATURE_SERVER_REGISTRY_TTL\nfrom feast.data_source import PushMode\nfrom feast.errors import PushSourceNotFoundException\n\n\n# TODO: deprecate this in favor of push features\nclass WriteToFeatureStoreRequest(BaseModel):\n    feature_view_name: str\n    df: dict\n    allow_registry_cache: bool = True\n\n\nclass PushFeaturesRequest(BaseModel):\n    push_source_name: str\n    df: dict\n    allow_registry_cache: bool = True\n    to: str = \"online\"\n\n\nclass MaterializeRequest(BaseModel):\n    start_ts: str\n    end_ts: str\n    feature_views: Optional[List[str]] = None\n\n\nclass MaterializeIncrementalRequest(BaseModel):\n    end_ts: str\n    feature_views: Optional[List[str]] = None\n\n\ndef get_app(\n    store: \"feast.FeatureStore\",\n    registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,\n):\n    proto_json.patch()\n\n    app = FastAPI()\n    # Asynchronously refresh registry, notifying shutdown and canceling the active timer if the app is shutting down\n    registry_proto = None\n    shutting_down = False\n    active_timer: Optional[threading.Timer] = None\n\n    async def get_body(request: Request):\n        return await request.body()\n\n    def async_refresh():\n        store.refresh_registry()\n        nonlocal registry_proto\n        registry_proto = store.registry.proto()\n        if shutting_down:\n            return\n        nonlocal active_timer\n        active_timer = threading.Timer(registry_ttl_sec, async_refresh)\n        active_timer.start()\n\n    @app.on_event(\"shutdown\")\n    def shutdown_event():\n        nonlocal shutting_down\n        shutting_down = True\n        if active_timer:\n            active_timer.cancel()\n\n    async_refresh()\n\n    @app.post(\"/get-online-features\")\n    def get_online_features(body=Depends(get_body)):\n        try:\n            body = json.loads(body)\n            # Initialize parameters for FeatureStore.get_online_features(...) call\n            if \"feature_service\" in body:\n                features = store.get_feature_service(\n                    body[\"feature_service\"], allow_cache=True\n                )\n            else:\n                features = body[\"features\"]\n\n            full_feature_names = body.get(\"full_feature_names\", False)\n\n            response_proto = store._get_online_features(\n                features=features,\n                entity_values=body[\"entities\"],\n                full_feature_names=full_feature_names,\n            ).proto\n\n            # Convert the Protobuf object to JSON and return it\n            return MessageToDict(\n                response_proto, preserving_proto_field_name=True, float_precision=18\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/push\")\n    def push(body=Depends(get_body)):\n        try:\n            request = PushFeaturesRequest(**json.loads(body))\n            df = pd.DataFrame(request.df)\n            if request.to == \"offline\":\n                to = PushMode.OFFLINE\n            elif request.to == \"online\":\n                to = PushMode.ONLINE\n            elif request.to == \"online_and_offline\":\n                to = PushMode.ONLINE_AND_OFFLINE\n            else:\n                raise ValueError(\n                    f\"{request.to} is not a supported push format. Please specify one of these ['online', 'offline', 'online_and_offline'].\"\n                )\n            store.push(\n                push_source_name=request.push_source_name,\n                df=df,\n                allow_registry_cache=request.allow_registry_cache,\n                to=to,\n            )\n        except PushSourceNotFoundException as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=422, detail=str(e))\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/write-to-online-store\")\n    def write_to_online_store(body=Depends(get_body)):\n        warnings.warn(\n            \"write_to_online_store is deprecated. Please consider using /push instead\",\n            RuntimeWarning,\n        )\n        try:\n            request = WriteToFeatureStoreRequest(**json.loads(body))\n            df = pd.DataFrame(request.df)\n            store.write_to_online_store(\n                feature_view_name=request.feature_view_name,\n                df=df,\n                allow_registry_cache=request.allow_registry_cache,\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.get(\"/health\")\n    def health():\n        return Response(status_code=status.HTTP_200_OK)\n\n    @app.post(\"/materialize\")\n    def materialize(body=Depends(get_body)):\n        try:\n            request = MaterializeRequest(**json.loads(body))\n            store.materialize(\n                utils.make_tzaware(parser.parse(request.start_ts)),\n                utils.make_tzaware(parser.parse(request.end_ts)),\n                request.feature_views,\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    @app.post(\"/materialize-incremental\")\n    def materialize_incremental(body=Depends(get_body)):\n        try:\n            request = MaterializeIncrementalRequest(**json.loads(body))\n            store.materialize_incremental(\n                utils.make_tzaware(parser.parse(request.end_ts)), request.feature_views\n            )\n        except Exception as e:\n            # Print the original exception on the server side\n            logger.exception(traceback.format_exc())\n            # Raise HTTPException to return the error message to the client\n            raise HTTPException(status_code=500, detail=str(e))\n\n    return app\n\n\nif sys.platform != \"win32\":\n    import gunicorn.app.base\n\n    class FeastServeApplication(gunicorn.app.base.BaseApplication):\n        def __init__(self, store: \"feast.FeatureStore\", **options):\n            self._app = get_app(\n                store=store,\n                registry_ttl_sec=options[\"registry_ttl_sec\"],\n            )\n            self._options = options\n            super().__init__()\n\n        def load_config(self):\n            for key, value in self._options.items():\n                if key.lower() in self.cfg.settings and value is not None:\n                    self.cfg.set(key.lower(), value)\n\n            self.cfg.set(\"worker_class\", \"uvicorn.workers.UvicornWorker\")\n\n        def load(self):\n            return self._app\n\n\ndef start_server(\n    store: \"feast.FeatureStore\",\n    host: str,\n    port: int,\n    no_access_log: bool,\n    workers: int,\n    keep_alive_timeout: int,\n    registry_ttl_sec: int,\n):\n    if sys.platform != \"win32\":\n        FeastServeApplication(\n            store=store,\n            bind=f\"{host}:{port}\",\n            accesslog=None if no_access_log else \"-\",\n            workers=workers,\n            keepalive=keep_alive_timeout,\n            registry_ttl_sec=registry_ttl_sec,\n        ).run()\n    else:\n        import uvicorn\n\n        app = get_app(store, registry_ttl_sec)\n        uvicorn.run(app, host=host, port=port, access_log=(not no_access_log))\n", "path": "sdk/python/feast/feature_server.py"}]}