| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent dda43c2 commit 9902064
3 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -40,7 +40,6 @@ | |||
| 40 | 40 | from fastapi.logger import logger | |
| 41 | 41 | from fastapi.responses import JSONResponse | |
| 42 | 42 | from fastapi.staticfiles import StaticFiles | |
| 43 | - from google.protobuf.json_format import MessageToDict | ||
| 44 | 43 | from pydantic import BaseModel, field_validator | |
| 45 | 44 | ||
| 46 | 45 | import feast | |
@@ -52,6 +51,7 @@ | |||
| 52 | 51 | FeastError, | |
| 53 | 52 | ) | |
| 54 | 53 | from feast.feast_object import FeastObject | |
| 54 | + from feast.feature_server_utils import convert_response_to_dict | ||
| 55 | 55 | from feast.feature_view_utils import get_feature_view_from_feature_store | |
| 56 | 56 | from feast.permissions.action import WRITE, AuthzedAction | |
| 57 | 57 | from feast.permissions.security_manager import assert_permissions | |
@@ -392,12 +392,9 @@ async def get_online_features(request: GetOnlineFeaturesRequest) -> Any: | |||
| 392 | 392 | ) | |
| 393 | 393 | ||
| 394 | 394 | response_dict = await run_in_threadpool( | |
| 395 | - MessageToDict, | ||
| 396 | - response.proto, | ||
| 397 | - preserving_proto_field_name=True, | ||
| 398 | - float_precision=18, | ||
| 395 | + convert_response_to_dict, response.proto | ||
| 399 | 396 | ) | |
| 400 | - return response_dict | ||
| 397 | + return JSONResponse(content=response_dict) | ||
| 401 | 398 | ||
| 402 | 399 | @app.post( | |
| 403 | 400 | "/retrieve-online-documents", | |
@@ -434,12 +431,9 @@ async def retrieve_online_documents( | |||
| 434 | 431 | ) | |
| 435 | 432 | ||
| 436 | 433 | response_dict = await run_in_threadpool( | |
| 437 | - MessageToDict, | ||
| 438 | - response.proto, | ||
| 439 | - preserving_proto_field_name=True, | ||
| 440 | - float_precision=18, | ||
| 434 | + convert_response_to_dict, response.proto | ||
| 441 | 435 | ) | |
| 442 | - return response_dict | ||
| 436 | + return JSONResponse(content=response_dict) | ||
| 443 | 437 | ||
| 444 | 438 | @app.post("/push", dependencies=[Depends(inject_user_details)]) | |
| 445 | 439 | async def push(request: PushFeaturesRequest) -> Response: | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,141 @@ | |||
| 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 | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments