159 lines
5.0 KiB
Python
159 lines
5.0 KiB
Python
import os
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import FastAPI, Response
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from fastapi.staticfiles import StaticFiles
|
|
from starlette.middleware.gzip import GZipMiddleware
|
|
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
|
|
|
|
from src.api.v1.router import api_router
|
|
from src.core.config import settings
|
|
from src.core.exception_handlers import register_exception_handlers
|
|
from src.db.base import engine
|
|
from src.db.session import create_tables
|
|
|
|
if not settings.DEBUG:
|
|
from raisa_fastapi_protected_api import (
|
|
AuthorizationMiddleware,
|
|
OpenEndpoint,
|
|
ProtectedOAuthSettings,
|
|
ProtectedRolesSettings,
|
|
ProtectedSettings,
|
|
SearchType,
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
if not settings.DEBUG:
|
|
await create_tables()
|
|
ProtectedSettings(
|
|
ProtectedOAuthSettings(JWKS_URI=settings.JWKS_URL),
|
|
ProtectedRolesSettings(
|
|
APP_NAMESPACE=settings.APP_NAMESPACE,
|
|
APP_NAME=settings.APP_NAME,
|
|
),
|
|
)
|
|
try:
|
|
yield
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
app = FastAPI(
|
|
title="DFiP Budget Planning API",
|
|
version="1.0.0",
|
|
lifespan=lifespan,
|
|
root_path=settings.ROOT_PATH,
|
|
)
|
|
register_exception_handlers(app)
|
|
|
|
VERSION = "1.0.0"
|
|
UP_TIME = datetime.now(timezone.utc)
|
|
|
|
|
|
@app.get("/healthcheck", status_code=200)
|
|
async def healthcheck():
|
|
return {"uptime": UP_TIME, "version": VERSION}
|
|
|
|
|
|
@app.get("/healthcheck2", status_code=200)
|
|
async def healthcheck2():
|
|
return {
|
|
"uptime": UP_TIME,
|
|
"version": VERSION,
|
|
"debug": settings.DEBUG,
|
|
"host": settings.HOST,
|
|
"port": settings.PORT,
|
|
}
|
|
|
|
@app.get("/healthz", status_code=200)
|
|
async def healthz():
|
|
return {"status": "ok", "uptime": UP_TIME, "version": VERSION}
|
|
|
|
|
|
@app.get("/readyz")
|
|
async def readyz(response: Response):
|
|
result = {"status": "ok", "db_status": "ok", "mv_expense_item_tree": "ok"}
|
|
try:
|
|
async with engine.connect() as conn:
|
|
await conn.execute(text("SELECT 1"))
|
|
if "postgresql" in settings.DATABASE_URL:
|
|
my_exists = (
|
|
await conn.execute(
|
|
text(
|
|
"""
|
|
SELECT 1
|
|
FROM pg_matviews
|
|
WHERE schemaname = 'v3'
|
|
AND matviewname = 'mv_expense_item_tree'
|
|
"""
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if my_exists is None:
|
|
result["status"] = "fail"
|
|
result["mv_expense_item_tree"] = "missing"
|
|
response.status_code = 503
|
|
except SQLAlchemyError as exc:
|
|
result["status"] = "fail"
|
|
result["db"] = f"error: {exc.__class__.__name__}"
|
|
response.status_code = 503
|
|
return result
|
|
|
|
|
|
|
|
if not settings.DEBUG:
|
|
app.add_middleware(
|
|
AuthorizationMiddleware,
|
|
open_endpoints=[
|
|
OpenEndpoint(path="", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(path=settings.ROOT_PATH + "", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(path="/", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(path=settings.ROOT_PATH + "/", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(path="/healthcheck", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(
|
|
path=settings.ROOT_PATH + "/healthcheck",
|
|
type_search=SearchType.ABSOLUTE,
|
|
),
|
|
OpenEndpoint(path="/healthz", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(
|
|
path=settings.ROOT_PATH + "/healthz",
|
|
type_search=SearchType.ABSOLUTE,
|
|
),
|
|
OpenEndpoint(path="/readyz", type_search=SearchType.ABSOLUTE),
|
|
OpenEndpoint(
|
|
path=settings.ROOT_PATH + "/readyz",
|
|
type_search=SearchType.ABSOLUTE,
|
|
),
|
|
OpenEndpoint(path="/openapi.json", type_search=SearchType.START),
|
|
OpenEndpoint(
|
|
path=settings.ROOT_PATH + "/openapi.json",
|
|
type_search=SearchType.START,
|
|
),
|
|
OpenEndpoint(path="/docs", type_search=SearchType.START),
|
|
OpenEndpoint(
|
|
path=settings.ROOT_PATH + "/docs",
|
|
type_search=SearchType.START,
|
|
),
|
|
],
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])
|
|
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1)
|
|
|
|
os.makedirs("web", exist_ok=True)
|
|
app.include_router(api_router, prefix="/api/v1")
|
|
app.mount("/", StaticFiles(directory="web", html=True), name="web")
|