97 lines
2.6 KiB
Python
97 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from fastapi import FastAPI, Request
|
|
from sqladmin import Admin
|
|
from starlette.responses import RedirectResponse
|
|
|
|
from app.admin.routes import router as admin_router
|
|
from app.admin.views import (
|
|
AuditLogAdmin,
|
|
JobAdmin,
|
|
JobLogAdmin,
|
|
LdapGroupAdmin,
|
|
LdapGroupRoleAdmin,
|
|
PermissionAdmin,
|
|
RoleAdmin,
|
|
UserAdmin,
|
|
)
|
|
from app.core.config import settings
|
|
from app.core.logging import setup_logging
|
|
from app.db.engine import engine
|
|
from app.db.schema import ensure_schema
|
|
from app.security.bootstrap import bootstrap_admin
|
|
from app.security.session import get_current_user
|
|
from app.security.fernet import get_or_create_fernet_key
|
|
from app.api.auth_routes import router as auth_router
|
|
|
|
|
|
def _init_db() -> None:
|
|
ensure_schema(engine)
|
|
|
|
|
|
def _ensure_runtime() -> None:
|
|
# 确保 data 目录存在
|
|
os.makedirs(settings.data_dir, exist_ok=True)
|
|
if settings.log_dir:
|
|
os.makedirs(settings.log_dir, exist_ok=True)
|
|
# 确保 Fernet key 准备好（或自动生成）
|
|
get_or_create_fernet_key(settings.fernet_key_path)
|
|
_init_db()
|
|
bootstrap_admin()
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
setup_logging()
|
|
_ensure_runtime()
|
|
|
|
app = FastAPI(title=settings.app_name)
|
|
|
|
app.include_router(auth_router)
|
|
app.include_router(admin_router)
|
|
|
|
admin = Admin(app=app, engine=engine, templates_dir="app/admin/templates")
|
|
admin.add_view(JobAdmin)
|
|
admin.add_view(JobLogAdmin)
|
|
admin.add_view(UserAdmin)
|
|
admin.add_view(RoleAdmin)
|
|
admin.add_view(PermissionAdmin)
|
|
admin.add_view(LdapGroupAdmin)
|
|
admin.add_view(LdapGroupRoleAdmin)
|
|
admin.add_view(AuditLogAdmin)
|
|
|
|
@app.middleware("http")
|
|
async def auth_middleware(request: Request, call_next):
|
|
if not settings.auth_enabled:
|
|
return await call_next(request)
|
|
|
|
path = request.url.path
|
|
public_prefixes = ("/login", "/logout", "/health")
|
|
static_prefixes = ("/admin/static", "/static")
|
|
if path.startswith(public_prefixes) or path.startswith(static_prefixes):
|
|
return await call_next(request)
|
|
|
|
user = get_current_user(request)
|
|
if not user:
|
|
next_url = request.url.path
|
|
if request.url.query:
|
|
next_url = f"{next_url}?{request.url.query}"
|
|
return RedirectResponse(f"/login?next={next_url}", status_code=303)
|
|
return await call_next(request)
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"ok": True, "name": settings.app_name}
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|