Mount the REST registry API routes on the UI server under /api/v1.
(app: FastAPI, store: "feast.FeatureStore")
| 74 | |
| 75 | |
| 76 | def _setup_rest_mode(app: FastAPI, store: "feast.FeatureStore"): |
| 77 | """Mount the REST registry API routes on the UI server under /api/v1.""" |
| 78 | from feast.api.registry.rest import register_all_routes |
| 79 | from feast.registry_server import RegistryServer |
| 80 | |
| 81 | grpc_handler = RegistryServer(store.registry) |
| 82 | |
| 83 | rest_app = FastAPI(root_path="/api/v1") |
| 84 | register_all_routes(rest_app, grpc_handler, store=store) |
| 85 | |
| 86 | class PushRequest(BaseModel): |
| 87 | push_source_name: str |
| 88 | df: Dict[str, List] |
| 89 | to: Optional[str] = "online" |
| 90 | |
| 91 | @rest_app.post("/push") |
| 92 | def push_labels(request: PushRequest): |
| 93 | """Push label data to a LabelView (or any PushSource-backed FeatureView).""" |
| 94 | try: |
| 95 | df = pd.DataFrame(request.df) |
| 96 | if "event_timestamp" in df.columns: |
| 97 | df["event_timestamp"] = pd.to_datetime( |
| 98 | df["event_timestamp"], utc=True |
| 99 | ).dt.tz_localize(None) |
| 100 | to = request.to or "online" |
| 101 | if to == "online_and_offline": |
| 102 | store.push( |
| 103 | request.push_source_name, |
| 104 | df, |
| 105 | to=feast.data_source.PushMode.ONLINE_AND_OFFLINE, |
| 106 | ) |
| 107 | elif to == "offline": |
| 108 | store.push( |
| 109 | request.push_source_name, df, to=feast.data_source.PushMode.OFFLINE |
| 110 | ) |
| 111 | else: |
| 112 | store.push( |
| 113 | request.push_source_name, df, to=feast.data_source.PushMode.ONLINE |
| 114 | ) |
| 115 | return {"status": "ok"} |
| 116 | except Exception: |
| 117 | logger.exception("Push failed") |
| 118 | return _safe_error_response("Push") |
| 119 | |
| 120 | class GetOnlineFeaturesRequest(BaseModel): |
| 121 | feature_view: str |
| 122 | entity_keys: Dict[str, List] |
| 123 | |
| 124 | @rest_app.post("/get-online-labels") |
| 125 | def get_online_labels(request: GetOnlineFeaturesRequest): |
| 126 | """Retrieve current label values from the online store for given entity keys.""" |
| 127 | try: |
| 128 | fv_name = request.feature_view |
| 129 | |
| 130 | fv: Any = None |
| 131 | try: |
| 132 | fv = store.get_feature_view(fv_name) |
| 133 | except Exception: |
no test coverage detected