36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
"""Authenticated graceful shutdown endpoint for the UVM API server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
|
|
|
|
from ..api_models import ServerExitResponse
|
|
from ..app import Application
|
|
from .dependencies import get_authorized_application
|
|
|
|
|
|
router = APIRouter(tags=["server"])
|
|
|
|
|
|
@router.post("/exit", response_model=ServerExitResponse, status_code=status.HTTP_202_ACCEPTED)
|
|
def exit_server(
|
|
background_tasks: BackgroundTasks,
|
|
request: Request,
|
|
_application: Application = Depends(get_authorized_application),
|
|
) -> ServerExitResponse:
|
|
"""Stop the Uvicorn process after this response has been delivered."""
|
|
server = getattr(request.app.state, "uvicorn_server", None)
|
|
if server is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="server shutdown is unavailable outside uvm --serve",
|
|
)
|
|
background_tasks.add_task(_request_server_exit, server)
|
|
return ServerExitResponse()
|
|
|
|
|
|
def _request_server_exit(server: Any) -> None:
|
|
server.should_exit = True
|