|
| 1 | +"""Fast serialization utilities for Feature Server responses. |
| 2 | +
|
| 3 | +Matches the output format of MessageToDict with proto_json.patch() applied. |
| 4 | +Values are serialized as native Python types (not wrapped dicts). |
| 5 | +""" |
| 6 | + |
| 7 | +import base64 |
| 8 | +import logging |
| 9 | +from datetime import datetime, timezone |
| 10 | +from typing import Any, Dict, Optional |
| 11 | + |
| 12 | +from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesResponse |
| 13 | +from feast.protos.feast.types.Value_pb2 import Value |
| 14 | + |
| 15 | +logger = logging.getLogger(__name__) |
| 16 | + |
| 17 | +# FieldStatus enum mapping (protos/feast/serving/ServingService.proto) |
| 18 | +_STATUS_NAMES: Dict[int, str] = { |
| 19 | + 0: "INVALID", |
| 20 | + 1: "PRESENT", |
| 21 | + 2: "NULL_VALUE", |
| 22 | + 3: "NOT_FOUND", |
| 23 | + 4: "OUTSIDE_MAX_AGE", |
| 24 | +} |
| 25 | + |
| 26 | + |
| 27 | +def convert_response_to_dict(response: GetOnlineFeaturesResponse) -> Dict[str, Any]: |
| 28 | + """Convert GetOnlineFeaturesResponse to a JSON-serializable dict. |
| 29 | +
|
| 30 | + Matches the structure produced by MessageToDict(proto, preserving_proto_field_name=True) |
| 31 | + with proto_json.patch() applied, with one intentional difference: |
| 32 | +
|
| 33 | + - double_val fields are returned as Python float objects (json.dumps uses Python 3.1+ |
| 34 | + shortest round-trip form, ~15-17 sig digits) rather than 18 fixed significant digits |
| 35 | + (float_precision=18). Values are numerically identical; only the JSON string length |
| 36 | + may differ. This is safe for all ML feature types and avoids unnecessary precision |
| 37 | + overhead. |
| 38 | + """ |
| 39 | + result: Dict[str, Any] = { |
| 40 | + "results": [ |
| 41 | + { |
| 42 | + "values": [_value_to_native(v) for v in feature_vector.values], |
| 43 | + "statuses": [ |
| 44 | + _STATUS_NAMES.get(s, "INVALID") for s in feature_vector.statuses |
| 45 | + ], |
| 46 | + **( |
| 47 | + { |
| 48 | + "event_timestamps": [ |
| 49 | + _timestamp_to_str(ts) |
| 50 | + for ts in feature_vector.event_timestamps |
| 51 | + ] |
| 52 | + } |
| 53 | + if feature_vector.event_timestamps |
| 54 | + else {} |
| 55 | + ), |
| 56 | + } |
| 57 | + for feature_vector in response.results |
| 58 | + ] |
| 59 | + } |
| 60 | + |
| 61 | + if response.HasField("metadata"): |
| 62 | + result["metadata"] = _metadata_to_dict(response.metadata) |
| 63 | + |
| 64 | + if response.status: |
| 65 | + result["status"] = response.status |
| 66 | + |
| 67 | + return result |
| 68 | + |
| 69 | + |
| 70 | +def _value_to_native(v: Value) -> Optional[Any]: |
| 71 | + """Convert a Value proto to a JSON-serializable Python type. |
| 72 | +
|
| 73 | + bytes_val and bytes_list_val are base64-encoded (RFC 4648) so that |
| 74 | + JSONResponse can serialize them without TypeError. This matches standard |
| 75 | + protobuf JSON encoding for bytes fields and is safe for all HTTP clients. |
| 76 | + """ |
| 77 | + which = v.WhichOneof("val") |
| 78 | + if which is None or which == "null_val": |
| 79 | + return None |
| 80 | + # bytes must be base64-encoded for JSON serialization |
| 81 | + elif which == "bytes_val": |
| 82 | + return base64.b64encode(v.bytes_val).decode("ascii") |
| 83 | + # RepeatedValue — nested Values that must be recursively converted |
| 84 | + elif which in ("list_val", "set_val"): |
| 85 | + return [_value_to_native(nested) for nested in getattr(v, which).val] |
| 86 | + # Map<string, Value> — recursively convert nested Values |
| 87 | + elif which in ("map_val", "struct_val"): |
| 88 | + return {k: _value_to_native(vv) for k, vv in getattr(v, which).val.items()} |
| 89 | + # MapList — list of Map<string, Value> |
| 90 | + elif which in ("map_list_val", "struct_list_val"): |
| 91 | + return [ |
| 92 | + {k: _value_to_native(vv) for k, vv in m.val.items()} |
| 93 | + for m in getattr(v, which).val |
| 94 | + ] |
| 95 | + # scalar_map_val has non-string keys; full conversion requires extra work and |
| 96 | + # this type is not returned by standard get_online_features paths today. |
| 97 | + elif which == "scalar_map_val": |
| 98 | + logger.warning( |
| 99 | + "scalar_map_val is not yet supported by convert_response_to_dict; value will be None" |
| 100 | + ) |
| 101 | + return None |
| 102 | + # bytes_list_val / bytes_set_val — base64-encode each element |
| 103 | + elif which in ("bytes_list_val", "bytes_set_val"): |
| 104 | + return [base64.b64encode(b).decode("ascii") for b in getattr(v, which).val] |
| 105 | + # All other list/set types have scalar .val fields |
| 106 | + elif "_list_" in which or "_set_" in which: |
| 107 | + return list(getattr(v, which).val) |
| 108 | + else: |
| 109 | + return getattr(v, which) |
| 110 | + |
| 111 | + |
| 112 | +def _timestamp_to_str(ts) -> str: |
| 113 | + """Convert protobuf Timestamp to RFC 3339 format with Z suffix. |
| 114 | +
|
| 115 | + Uses adaptive precision to match MessageToDict output: |
| 116 | + - No fractional seconds when nanos == 0 |
| 117 | + - 3 digits (milliseconds) when nanos % 1_000_000 == 0 |
| 118 | + - 6 digits (microseconds) when nanos % 1_000 == 0 |
| 119 | + - 9 digits (nanoseconds) otherwise |
| 120 | + """ |
| 121 | + if ts.seconds == 0 and ts.nanos == 0: |
| 122 | + return "1970-01-01T00:00:00Z" |
| 123 | + dt = datetime.fromtimestamp(ts.seconds, tz=timezone.utc) |
| 124 | + base = dt.strftime("%Y-%m-%dT%H:%M:%S") |
| 125 | + nanos = ts.nanos |
| 126 | + if nanos == 0: |
| 127 | + return base + "Z" |
| 128 | + elif nanos % 1_000_000 == 0: |
| 129 | + return base + ".%03dZ" % (nanos // 1_000_000) |
| 130 | + elif nanos % 1_000 == 0: |
| 131 | + return base + ".%06dZ" % (nanos // 1_000) |
| 132 | + else: |
| 133 | + return base + ".%09dZ" % nanos |
| 134 | + |
| 135 | + |
| 136 | +def _metadata_to_dict(metadata) -> Dict[str, Any]: |
| 137 | + """Convert FeatureResponseMeta to dict (matches proto_json.patch() format).""" |
| 138 | + result: Dict[str, Any] = {} |
| 139 | + if metadata.HasField("feature_names"): |
| 140 | + result["feature_names"] = list(metadata.feature_names.val) |
| 141 | + return result |
0 commit comments