68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
|
"""
|
||
|
|
Application root file: Inits the FastAPI application
|
||
|
|
"""
|
||
|
|
from contextlib import asynccontextmanager
|
||
|
|
from typing import AsyncGenerator
|
||
|
|
from prometheus_client import make_asgi_app
|
||
|
|
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from starlette.middleware.sessions import SessionMiddleware
|
||
|
|
from starlette.middleware.cors import CORSMiddleware
|
||
|
|
|
||
|
|
from src.config import settings
|
||
|
|
from src.api import api_router
|
||
|
|
from src.prometheus import prometheus
|
||
|
|
|
||
|
|
from src.auth.config import auth_settings
|
||
|
|
from src.misp.service import MISPHandler
|
||
|
|
|
||
|
|
# TODO: Create Pydantic request/response schemas
|
||
|
|
|
||
|
|
|
||
|
|
@asynccontextmanager
|
||
|
|
async def lifespan(_application: FastAPI) -> AsyncGenerator:
|
||
|
|
# Startup
|
||
|
|
yield
|
||
|
|
# Shutdown
|
||
|
|
_application.misp_handler.stop_timer()
|
||
|
|
|
||
|
|
|
||
|
|
if settings.ENVIRONMENT.is_deployed:
|
||
|
|
# Do this only on prod
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
app = FastAPI(
|
||
|
|
swagger_ui_init_oauth={
|
||
|
|
"clientId": auth_settings.CLIENT_ID,
|
||
|
|
"usePkceWithAuthorizationCodeGrant": True,
|
||
|
|
"scopes": "openid profile email",
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
metrics_app = make_asgi_app()
|
||
|
|
app.mount("/metrics", metrics_app)
|
||
|
|
|
||
|
|
prometheus.APP_STATE.state("starting")
|
||
|
|
|
||
|
|
# Type inspection disabled for middleware injection.
|
||
|
|
# Known bug in FastAPI type checking: https://github.com/astral-sh/ty/issues/1635
|
||
|
|
# noinspection PyTypeChecker
|
||
|
|
app.add_middleware(SessionMiddleware, secret_key=settings.SECRET_KEY.get_secret_value())
|
||
|
|
# noinspection PyTypeChecker
|
||
|
|
app.add_middleware(
|
||
|
|
CORSMiddleware,
|
||
|
|
allow_origins=settings.CORS_ORIGINS,
|
||
|
|
allow_origin_regex=settings.CORS_ORIGINS_REGEX,
|
||
|
|
allow_credentials=True,
|
||
|
|
allow_methods=("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"),
|
||
|
|
allow_headers=settings.CORS_HEADERS,
|
||
|
|
)
|
||
|
|
|
||
|
|
app.include_router(api_router)
|
||
|
|
|
||
|
|
app.misp_handler = MISPHandler()
|
||
|
|
|
||
|
|
prometheus.APP_STATE.state("running")
|