import json
import threading
from typing import Callable, Optional
import importlib_resources
import uvicorn
from fastapi import FastAPI, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import feast
[docs]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(__name__) / "ui/build/"
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():
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 + "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
[docs]def start_server(
store: "feast.FeatureStore",
host: str,
port: int,
get_registry_dump: Callable,
project_id: str,
registry_ttl_sec: int,
root_path: str = "",
):
app = get_app(
store,
project_id,
registry_ttl_sec,
root_path,
)
uvicorn.run(app, host=host, port=port)