-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathui_server.py
More file actions
124 lines (106 loc) · 3.35 KB
/
ui_server.py
File metadata and controls
124 lines (106 loc) · 3.35 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
import json
import threading
from importlib import resources as importlib_resources
from typing import Callable, Optional
import uvicorn
from fastapi import FastAPI, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import feast
def get_app(
store: "feast.FeatureStore",
project_id: str,
registry_ttl_secs: int,
root_path: str = "",
):
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Asynchronously refresh registry, notifying shutdown and canceling the active timer if the app is shutting down
registry_proto = None
shutting_down = False
active_timer: Optional[threading.Timer] = None
def async_refresh():
store.refresh_registry()
nonlocal registry_proto
registry_proto = store.registry.proto()
if shutting_down:
return
nonlocal active_timer
active_timer = threading.Timer(registry_ttl_secs, async_refresh)
active_timer.start()
@app.on_event("shutdown")
def shutdown_event():
nonlocal shutting_down
shutting_down = True
if active_timer:
active_timer.cancel()
async_refresh()
ui_dir_ref = importlib_resources.files(__spec__.parent) / "ui/build/" # type: ignore[name-defined, arg-type]
with importlib_resources.as_file(ui_dir_ref) as ui_dir:
# Initialize with the projects-list.json file
with ui_dir.joinpath("projects-list.json").open(mode="w") as f:
projects_dict = {
"projects": [
{
"name": "Project",
"description": "Test project",
"id": project_id,
"registryPath": f"{root_path}/registry",
}
]
}
f.write(json.dumps(projects_dict))
@app.get("/registry")
def read_registry():
if registry_proto is None:
return Response(status_code=503) # Service Unavailable
return Response(
content=registry_proto.SerializeToString(),
media_type="application/octet-stream",
)
# For all other paths (such as paths that would otherwise be handled by react router), pass to React
@app.api_route("/p/{path_name:path}", methods=["GET"])
def catch_all():
filename = ui_dir.joinpath("index.html")
with open(filename) as f:
content = f.read()
return Response(content, media_type="text/html")
app.mount(
"/",
StaticFiles(directory=ui_dir, html=True),
name="site",
)
return app
def start_server(
store: "feast.FeatureStore",
host: str,
port: int,
get_registry_dump: Callable,
project_id: str,
registry_ttl_sec: int,
root_path: str = "",
tls_key_path: str = "",
tls_cert_path: str = "",
):
app = get_app(
store,
project_id,
registry_ttl_sec,
root_path,
)
if tls_key_path and tls_cert_path:
uvicorn.run(
app,
host=host,
port=port,
ssl_keyfile=tls_key_path,
ssl_certfile=tls_cert_path,
)
else:
uvicorn.run(app, host=host, port=port)