55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
import inspect
|
|
import logging
|
|
import threading
|
|
from typing import Callable
|
|
|
|
import uvicorn
|
|
from dynaconf import Dynaconf
|
|
from fastapi import FastAPI
|
|
|
|
from pyinfra.config.loader import validate_settings
|
|
from pyinfra.config.validators import webserver_validators
|
|
|
|
|
|
def create_webserver_thread_from_settings(app: FastAPI, settings: Dynaconf) -> threading.Thread:
|
|
validate_settings(settings, validators=webserver_validators)
|
|
return create_webserver_thread(app=app, port=settings.webserver.port, host=settings.webserver.host)
|
|
|
|
|
|
def create_webserver_thread(app: FastAPI, port: int, host: str) -> threading.Thread:
|
|
"""Creates a thread that runs a FastAPI webserver. Start with thread.start(), and join with thread.join().
|
|
Note that the thread is a daemon thread, so it will be terminated when the main thread is terminated.
|
|
"""
|
|
thread = threading.Thread(target=lambda: uvicorn.run(app, port=port, host=host, log_level=logging.WARNING))
|
|
thread.daemon = True
|
|
return thread
|
|
|
|
|
|
HealthFunction = Callable[[], bool]
|
|
|
|
|
|
def add_health_check_endpoint(app: FastAPI, health_function: HealthFunction) -> FastAPI:
|
|
"""Add a health check endpoint to the app. The health function should return True if the service is healthy,
|
|
and False otherwise. The health function is called when the endpoint is hit.
|
|
"""
|
|
if inspect.iscoroutinefunction(health_function):
|
|
|
|
@app.get("/health")
|
|
@app.get("/ready")
|
|
async def async_check_health():
|
|
alive = await health_function()
|
|
if alive:
|
|
return {"status": "OK"}, 200
|
|
return {"status": "Service Unavailable"}, 503
|
|
|
|
else:
|
|
|
|
@app.get("/health")
|
|
@app.get("/ready")
|
|
def check_health():
|
|
if health_function():
|
|
return {"status": "OK"}, 200
|
|
return {"status": "Service Unavailable"}, 503
|
|
|
|
return app
|