Add native types for Python SDK online retrieval

sdk/python/feast/client.py

@@ -22,7 +22,8 @@
 import uuid
 from collections import OrderedDict
 from math import ceil
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
+import warnings
 
 import grpc
 import pandas as pd
@@ -83,15 +84,19 @@
     GetFeastServingInfoRequest,
     GetFeastServingInfoResponse,
     GetOnlineFeaturesRequest,
-    GetOnlineFeaturesResponse,
 )
 from feast.serving.ServingService_pb2_grpc import ServingServiceStub
 from tensorflow_metadata.proto.v0 import statistics_pb2
+from feast.type_map import python_type_to_feast_value_type, _python_value_to_proto_value
+from feast.response import OnlineResponse
+
 
 _logger = logging.getLogger(__name__)
 
 CPU_COUNT = os.cpu_count()  # type: int
 
+warnings.simplefilter("once", DeprecationWarning)
+
 
 class Client:
     """
@@ -655,10 +660,10 @@ def get_batch_features(
     def get_online_features(
         self,
         feature_refs: List[str],
-        entity_rows: List[GetOnlineFeaturesRequest.EntityRow],
+        entity_rows: List[Union[GetOnlineFeaturesRequest.EntityRow, Dict[str, Any]]],
         project: Optional[str] = None,
         omit_entities: bool = False,
-    ) -> GetOnlineFeaturesResponse:
+    ) -> OnlineResponse:
         """
         Retrieves the latest online feature data from Feast Serving
 
@@ -668,9 +673,13 @@ def get_online_features(
                 "feature_set:feature" where "feature_set" & "feature" refer to
                 the feature and feature set names respectively.
                 Only the feature name is required.
-            entity_rows: List of GetFeaturesRequest.EntityRow where each row
+            entity_rows:
+                List of GetFeaturesRequest.EntityRow where each row
                 contains entities. Timestamp should not be set for online
                 retrieval. All entity types within a feature
+                OR
+                List of Dict[str, Union[bool,bytes,float,int,str,List[bool,bytes,float,int,str]]]]
+                where each key represents the entity name and value is feast.types.Value in Python native form.
             project: Optionally specify the the project override. If specified, uses given project for retrieval.
                 Overrides the projects specified in Feature References if also are specified.
             omit_entities: If true will omit entity values in the returned feature data.
@@ -679,8 +688,13 @@ def get_online_features(
             Each EntityRow provided will yield one record, which contains
             data fields with data value and field status metadata (if included).
         """
-
+        warnings.warn(
+            "entity_rows parameter will only be accepting Dict format from v0.7 onwards",
+            DeprecationWarning,
+        )
         try:
+            if entity_rows and isinstance(entity_rows[0], dict):
+                entity_rows = _infer_entity_rows(entity_rows)
             response = self._serving_service.GetOnlineFeatures(
                 GetOnlineFeaturesRequest(
                     omit_entities_in_response=omit_entities,
@@ -692,6 +706,8 @@ def get_online_features(
         except grpc.RpcError as e:
             raise grpc.RpcError(e.details())
 
+        response = OnlineResponse(response)
+
         return response
 
     def list_ingest_jobs(
@@ -960,6 +976,41 @@ def _get_grpc_metadata(self):
         return ()
 
 
+def _infer_entity_rows(
+    entities: List[Dict[str, Any]]
+) -> List[GetOnlineFeaturesRequest.EntityRow]:
+    """
+    Builds a list of EntityRow protos from Python native type format passed by user.
+
+    Args:
+        entities: List of Dict[str, Union[bool,bytes,float,int,str,List[bool,bytes,float,int,str]]]]
+            where each key represents the entity name and value is feast.types.Value in Python native form.
+
+    Returns:
+        A list of EntityRow protos parsed from args.
+    """
+    entity_row_list = []
+    entity_type_map = dict()
+
+    for entity in entities:
+        for key, value in entity.items():
+            # Infer the specific type for this row
+            current_dtype = python_type_to_feast_value_type(name=key, value=value)
+
+            if key not in entity_type_map:
+                entity_type_map[key] = current_dtype
+            else:
+                if current_dtype != entity_type_map[key]:
+                    raise TypeError(
+                        f"Input entity {key} has mixed types and that is not allowed. "
+                    )
+            proto_value = _python_value_to_proto_value(current_dtype, value)
+            entity_row_list.append(
+                GetOnlineFeaturesRequest.EntityRow(fields={key: proto_value})
+            )
+    return entity_row_list
+
+
 def _build_feature_references(
     feature_ref_strs: List[str], project: Optional[str] = None
 ) -> List[FeatureReference]:

sdk/python/feast/response.py

@@ -0,0 +1,54 @@
+# Copyright 2020 The Feast Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Any, Dict
+
+from feast.serving.ServingService_pb2 import GetOnlineFeaturesResponse
+from feast.type_map import feast_value_type_to_python_type
+
+
+class OnlineResponse:
+    """
+    Defines a online response in feast.
+    """
+
+    def __init__(self, online_response_proto: GetOnlineFeaturesResponse):
+        """
+        Construct a native online response from its protobuf version.
+
+        Args:
+        online_response_proto: GetOnlineResponse proto object to construct from.
+        """
+        self.proto = online_response_proto
+
+    @property
+    def field_values(self):
+        """
+        Getter for GetOnlineResponse's field_values.
+        """
+        return self.proto.field_values
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Converts GetOnlineFeaturesResponse features into a dictionary form.
+        """
+        fields = [k for row in self.field_values for k, _ in row.fields.items()]
+        features_dict = {k: list() for k in fields}
+
+        for row in self.field_values:
+            for feature in features_dict.keys():
+                native_type_value = feast_value_type_to_python_type(row.fields[feature])
+                features_dict[feature].append(native_type_value)
+
+        return features_dict

sdk/python/feast/type_map.py

@@ -13,12 +13,13 @@
 # limitations under the License.
 
 from datetime import datetime, timezone
-from typing import List
+from typing import Any, List
 
 import numpy as np
 import pandas as pd
 import pyarrow as pa
 from google.protobuf.timestamp_pb2 import Timestamp
+from google.protobuf.json_format import MessageToDict
 from pyarrow.lib import TimestampType
 
 from feast.constants import DATETIME_COLUMN
@@ -38,6 +39,44 @@
 from feast.value_type import ValueType
 
 
+def feast_value_type_to_python_type(field_value_proto: ProtoValue) -> Any:
+    """
+    Converts field value Proto to Dict and returns each field's Feast Value Type value
+    in their respective Python value.
+
+    Args:
+        field_value_proto: Field value Proto
+
+    Returns:
+        Python native type representation/version of the given field_value_proto
+    """
+    field_value_dict = MessageToDict(field_value_proto)
+
+    for k, v in field_value_dict.items():
+        if k == "int64Val":
+            return int(v)
+        if k == "bytesVal":
+            return bytes(v)
+        if (k == "int64ListVal") or (k == "int32ListVal"):
+            return [int(item) for item in v["val"]]
+        if (k == "floatListVal") or (k == "doubleListVal"):
+            return [float(item) for item in v["val"]]
+        if k == "stringListVal":
+            return [str(item) for item in v["val"]]
+        if k == "bytesListVal":
+            return [bytes(item) for item in v["val"]]
+        if k == "boolListVal":
+            return [bool(item) for item in v["val"]]
+
+        if k in ["int32Val", "floatVal", "doubleVal", "stringVal", "boolVal"]:
+            return v
+        else:
+            raise TypeError(
+                f"Casting to Python native type for type {k} failed. "
+                f"Type {k} not found"
+            )
+
+
 def python_type_to_feast_value_type(
     name: str, value, recurse: bool = True
 ) -> ValueType:
@@ -80,6 +119,10 @@ def python_type_to_feast_value_type(
     if type_name in type_map:
         return type_map[type_name]
 
+    if isinstance(value, list):
+        type_name = "ndarray"
+        value = np.asarray(value)
+
     if type_name == "ndarray":
         if recurse: