feast/sdk/python/feast/feature_server_utils.py at master · Gwill/feast · GitHub

History

160 lines (136 loc) · 6.13 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

"""Fast serialization utilities for Feature Server responses.

Matches the output format of MessageToDict with proto_json.patch() applied.

Values are serialized as native Python types (not wrapped dicts).

"""

import base64

import logging

from datetime import datetime, timezone

from typing import Any, Dict, Optional

from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesResponse

from feast.protos.feast.types.Value_pb2 import Value

logger = logging.getLogger(__name__)

# FieldStatus enum mapping (protos/feast/serving/ServingService.proto)

_STATUS_NAMES: Dict[int, str] = {

0: "INVALID",

1: "PRESENT",

2: "NULL_VALUE",

3: "NOT_FOUND",

4: "OUTSIDE_MAX_AGE",

}

def convert_response_to_dict(response: GetOnlineFeaturesResponse) -> Dict[str, Any]:

"""Convert GetOnlineFeaturesResponse to a JSON-serializable dict.

Matches the structure produced by MessageToDict(proto, preserving_proto_field_name=True)

with proto_json.patch() applied, with one intentional difference:

- double_val fields are returned as Python float objects (json.dumps uses Python 3.1+

shortest round-trip form, ~15-17 sig digits) rather than 18 fixed significant digits

(float_precision=18). Values are numerically identical; only the JSON string length

may differ. This is safe for all ML feature types and avoids unnecessary precision

overhead.

"""

result: Dict[str, Any] = {

"results": [

{

"values": [_value_to_native(v) for v in feature_vector.values],

"statuses": [

_STATUS_NAMES.get(s, "INVALID") for s in feature_vector.statuses

**(

{

"event_timestamps": [

_timestamp_to_str(ts)

for ts in feature_vector.event_timestamps

]

}

if feature_vector.event_timestamps

else {}

}

for feature_vector in response.results

]

}

if response.HasField("metadata"):

result["metadata"] = _metadata_to_dict(response.metadata)

if response.status:

result["status"] = response.status

return result

def _value_to_native(v: Value) -> Optional[Any]:

"""Convert a Value proto to a JSON-serializable Python type.

bytes_val and bytes_list_val are base64-encoded (RFC 4648) so that

JSONResponse can serialize them without TypeError. This matches standard

protobuf JSON encoding for bytes fields and is safe for all HTTP clients.

"""

which = v.WhichOneof("val")

if which is None or which == "null_val":

return None

# bytes must be base64-encoded for JSON serialization

elif which == "bytes_val":

return base64.b64encode(v.bytes_val).decode("ascii")

# RepeatedValue — nested Values that must be recursively converted

elif which in ("list_val", "set_val"):

return [_value_to_native(nested) for nested in getattr(v, which).val]

# Map<string, Value> — recursively convert nested Values

elif which in ("map_val", "struct_val"):

return {k: _value_to_native(vv) for k, vv in getattr(v, which).val.items()}

# MapList — list of Map<string, Value>

elif which in ("map_list_val", "struct_list_val"):

return [

{k: _value_to_native(vv) for k, vv in m.val.items()}

for m in getattr(v, which).val

]

# scalar_map_val has non-string keys; full conversion requires extra work and

# this type is not returned by standard get_online_features paths today.

elif which == "scalar_map_val":

logger.warning(

"scalar_map_val is not yet supported by convert_response_to_dict; value will be None"

)

return None

# zoned_timestamp_val is a ZonedTimestamp message; serialize as an ISO 8601

# string in its stored zone so JSONResponse can encode it (the raw message

# is not JSON-serializable).

elif which == "zoned_timestamp_val":

return _zoned_timestamp_to_str(v.zoned_timestamp_val)

# bytes_list_val / bytes_set_val — base64-encode each element

elif which in ("bytes_list_val", "bytes_set_val"):

return [base64.b64encode(b).decode("ascii") for b in getattr(v, which).val]

# All other list/set types have scalar .val fields

elif "_list_" in which or "_set_" in which:

return list(getattr(v, which).val)

else:

return getattr(v, which)

def _zoned_timestamp_to_str(zoned) -> Optional[str]:

"""Convert a ZonedTimestamp proto to an ISO 8601 string in its stored zone.

A null instant (NaT sentinel) returns None. The zone is resolved via the

same logic used on the read path so IANA names and fixed offsets round-trip.

"""

from feast.type_map import NULL_TIMESTAMP_INT_VALUE, _zone_from_name

if zoned.unix_timestamp == NULL_TIMESTAMP_INT_VALUE:

return None

tz = _zone_from_name(zoned.zone)

return datetime.fromtimestamp(zoned.unix_timestamp, tz=tz).isoformat()

def _timestamp_to_str(ts) -> str:

"""Convert protobuf Timestamp to RFC 3339 format with Z suffix.

Uses adaptive precision to match MessageToDict output:

- No fractional seconds when nanos == 0

- 3 digits (milliseconds) when nanos % 1_000_000 == 0

- 6 digits (microseconds) when nanos % 1_000 == 0

- 9 digits (nanoseconds) otherwise

"""

if ts.seconds == 0 and ts.nanos == 0:

return "1970-01-01T00:00:00Z"

dt = datetime.fromtimestamp(ts.seconds, tz=timezone.utc)

base = dt.strftime("%Y-%m-%dT%H:%M:%S")

nanos = ts.nanos

if nanos == 0:

return base + "Z"

elif nanos % 1_000_000 == 0:

return base + ".%03dZ" % (nanos // 1_000_000)

elif nanos % 1_000 == 0:

return base + ".%06dZ" % (nanos // 1_000)

else:

return base + ".%09dZ" % nanos

def _metadata_to_dict(metadata) -> Dict[str, Any]:

"""Convert FeatureResponseMeta to dict (matches proto_json.patch() format)."""

result: Dict[str, Any] = {}

if metadata.HasField("feature_names"):

result["feature_names"] = list(metadata.feature_names.val)

return result

feast/sdk/python/feast/feature_server_utils.py at master · Gwill/feast · GitHub

Expand file tree

Breadcrumbs

feature_server_utils.py

Latest commit

History

Breadcrumbs

feature_server_utils.py

File metadata and controls