Add native types for Python SDK online retrieval

sdk/python/feast/client.py

@@ -22,7 +22,7 @@
 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 grpc
 import pandas as pd
@@ -87,6 +87,9 @@
 )
 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__)
 
@@ -655,10 +658,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 +671,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: Specifies the project which contain the FeatureSets
                 which the requested features belong to.
             omit_entities: If true will omit entity values in the returned feature data.
@@ -682,6 +689,8 @@ def get_online_features(
         """
 
         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,
@@ -725,6 +734,8 @@ def get_online_features(
         except grpc.RpcError as e:
             raise grpc.RpcError(e.details())
 
+        response = OnlineResponse(response)
+
         return response
 
     def list_ingest_jobs(
@@ -993,6 +1004,32 @@ 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 = []
+
+    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)
+            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.
+        """
+        features = [k for row in self.field_values for k, _ in row.fields.items()]
+        features_dict = dict.fromkeys(features)
+
+        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] = 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,36 @@
 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 based on Feast Value Type
+    """
+    field_value_dict = MessageToDict(field_value_proto)
+
+    for k, v in field_value_dict.items():
+        if k == "int64Val":
+            return int(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"]]
+
+        return v
+
+
 def python_type_to_feast_value_type(
     name: str, value, recurse: bool = True
 ) -> ValueType:

tests/e2e/redis/basic-ingest-redis-serving.py

@@ -741,6 +741,77 @@ def test_list_entities_and_features(client):
     assert set(filter_by_project_entity_expected) == set(filter_by_project_entity_actual)
     assert set(filter_by_project_labels_expected) == set(filter_by_project_labels_actual)
 
+
+@pytest.fixture(scope='module')
+def list_type_dataframe():
+    N_ROWS = 2
+    time_offset = datetime.utcnow().replace(tzinfo=pytz.utc)
+    customer_df = pd.DataFrame(
+        {
+            "datetime": [time_offset] * N_ROWS,
+            "customer_id": [i for i in range(N_ROWS)],
+            "rating": [i for i in range(N_ROWS)],
+            "cost": [float(i)+0.5 for i in range(N_ROWS)],
+            "past_transactions_int": [[i,i+2] for i in range(N_ROWS)],
+            "past_transactions_double": [[float(i)+0.5,float(i)+2] for i in range(N_ROWS)],
+            "past_transactions_float": [[float(i)+0.5,float(i)+2] for i in range(N_ROWS)],
+            "past_transactions_string": [['first_'+str(i),'second_'+str(i)] for i in range(N_ROWS)],
+            "past_transactions_bool": [[True,False] for _ in range(N_ROWS)]
+        }
+    )
+    return customer_df
+
+
+@pytest.mark.timeout(600)
+@pytest.mark.run(order=43)
+def test_basic_retrieve_online_dict(client, list_type_dataframe):
+    customer_fs = FeatureSet(
+        name="customer2",
+        features=[
+            Feature(name="rating", dtype=ValueType.INT64, labels={"key1":"val1"}),
+            Feature(name="cost", dtype=ValueType.FLOAT),
+            Feature(name="past_transactions_int", dtype=ValueType.INT64_LIST),
+            Feature(name="past_transactions_double", dtype=ValueType.DOUBLE_LIST),
+            Feature(name="past_transactions_float", dtype=ValueType.FLOAT_LIST),
+            Feature(name="past_transactions_string", dtype=ValueType.STRING_LIST),
+            Feature(name="past_transactions_bool", dtype=ValueType.BOOL_LIST)
+        ],
+        entities=[Entity("customer_id", ValueType.INT64)],
+        max_age=Duration(seconds=600)
+    )
+
+    client.set_project(PROJECT_NAME)
+    client.apply(customer_fs)
+
+    customer_fs = client.get_feature_set(name="customer2")
+    client.ingest(customer_fs, list_type_dataframe, timeout=300)
+    time.sleep(15)
+
+    online_request_entity = [{"customer_id": 1}]
+    online_request_features = [
+        "rating",
+        "cost",
+        "past_transactions_int",
+        "past_transactions_double",
+        "past_transactions_float",
+        "past_transactions_string",
+        "past_transactions_bool"
+    ]
+
+    online_features_actual = client.get_online_features(entity_rows=online_request_entity, feature_refs=online_request_features)
+    online_features_expected = {
+        "customer_id": 1,
+        "rating": 1,
+        "cost": 1.5,
+        "past_transactions_int": [1,3],
+        "past_transactions_double": [1.5,3.0],
+        "past_transactions_float": [1.5,3.0],
+        "past_transactions_string": ['first_1','second_1'],
+        "past_transactions_bool": [True,False]
+    }
+    assert online_features_actual.to_dict() == online_features_expected
+
+
 @pytest.mark.timeout(900)
 @pytest.mark.run(order=50)
 def test_sources_deduplicate_ingest_jobs(client):