Local feature server implementation (HTTP endpoint)

protos/feast/serving/ServingService.proto

@@ -32,6 +32,9 @@ service ServingService {
 
     // Get online features (v2) synchronously.
     rpc GetOnlineFeaturesV2 (GetOnlineFeaturesRequestV2) returns (GetOnlineFeaturesResponse);
+
+    // Get online features synchronously.
+    rpc GetOnlineFeatures (GetOnlineFeaturesRequest) returns (GetOnlineFeaturesResponse);
 }
 
 message GetFeastServingInfoRequest {}
@@ -81,6 +84,23 @@ message GetOnlineFeaturesRequestV2 {
     }
 }
 
+message FeatureListConfig {
+    repeated string features = 1;
+    bool full_feature_names = 2;
+}
+
+message EntityRow {
+    map<string,feast.types.Value> fields = 1;
+}
+
+message GetOnlineFeaturesRequest {
+    oneof features {
+        string feature_service_name = 1;
+        FeatureListConfig feature_list_config = 2;
+    }
+    repeated EntityRow entity_rows = 3;
+}
+
 message GetOnlineFeaturesResponse {
     // Feature values retrieved from feast.
     repeated FieldValues field_values = 1;

sdk/python/feast/cli.py

@@ -357,5 +357,19 @@ def init_command(project_directory, minimal: bool, template: str):
     init_repo(project_directory, template)
 
 
+@cli.command("serve")
+@click.option(
+    "--port", "-p", type=click.INT, default=6566, help="Specify a port for the server"
+)
+@click.pass_context
+def serve_command(ctx: click.Context, port: int):
+    """Start a the feature consumption server locally on a given port."""
+    repo = ctx.obj["CHDIR"]
+    cli_check_repo(repo)
+    store = FeatureStore(repo_path=str(repo))
+
+    store.serve(port)
+
+
 if __name__ == "__main__":
     cli()

sdk/python/feast/feature_server.py

@@ -0,0 +1,57 @@
+import uvicorn
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.logger import logger
+from google.protobuf.json_format import MessageToDict, Parse
+
+import feast
+from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest
+from feast.type_map import feast_value_type_to_python_type
+
+
+def get_app(store: "feast.FeatureStore"):
+    app = FastAPI()
+
+    @app.get("/get-online-features/")
+    async def get_online_features(request: Request):
+        try:
+            # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object
+            body = await request.body()
+            request_proto = GetOnlineFeaturesRequest()
+            Parse(body, request_proto)
+
+            # Initialize parameters for FeatureStore.get_online_features(...) call
+            if request_proto.HasField("feature_service_name"):
+                features = store.get_feature_service(request_proto.feature_service_name)
+                full_feature_names = False
+            else:
+                features = request_proto.feature_list_config.features
+                full_feature_names = (
+                    request_proto.feature_list_config.full_feature_names
+                )
+
+            entity_rows = [
+                {
+                    k: feast_value_type_to_python_type(v)
+                    for k, v in entity_row_proto.fields.items()
+                }
+                for entity_row_proto in request_proto.entity_rows
+            ]
+
+            response_proto = store.get_online_features(
+                features, entity_rows, full_feature_names=full_feature_names
+            ).proto
+
+            # Convert the Protobuf object to JSON and return it
+            return MessageToDict(response_proto, preserving_proto_field_name=True)
+        except Exception as e:
+            # Print the original exception on the server side
+            logger.exception(e)
+            # Raise HTTPException to return the error message to the client
+            raise HTTPException(status_code=500, detail=str(e))
+
+    return app
+
+
+def start_server(store: "feast.FeatureStore", port: int):
+    app = get_app(store)
+    uvicorn.run(app, host="127.0.0.1", port=port)

sdk/python/feast/feature_store.py

@@ -22,7 +22,7 @@
 from colorama import Fore, Style
 from tqdm import tqdm
 
-from feast import utils
+from feast import feature_server, utils
 from feast.entity import Entity
 from feast.errors import (
     EntityNotFoundException,
@@ -846,6 +846,11 @@ def get_online_features(
 
         return OnlineResponse(GetOnlineFeaturesResponse(field_values=result_rows))
 
+    @log_exceptions_and_usage
+    def serve(self, port: int) -> None:
+        """Start a the feature consumption server locally on a given port."""
+        feature_server.start_server(self, port)
+
 
 def _entity_row_to_key(row: GetOnlineFeaturesRequestV2.EntityRow) -> EntityKeyProto:
     names, values = zip(*row.fields.items())

sdk/python/setup.py

@@ -57,6 +57,8 @@
     "tenacity>=7.*",
     "toml==0.10.*",
     "tqdm==4.*",
+    "fastapi>=0.68.0",
+    "uvicorn[standard]>=0.14.0",
 ]
 
 GCP_REQUIRED = [