64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
import os
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import APIRouter, FastAPI
|
|
|
|
from api.assets import configure_static_assets
|
|
from banban.middleware.auth import install_auth_middleware
|
|
from banban.middleware.request_log import install_request_logging_middleware
|
|
from banban.routers.bindings import router as bindings_router
|
|
from banban.routers.children import router as children_router
|
|
from banban.routers.device_im import router as device_im_router
|
|
from banban.routers.devices import router as devices_router
|
|
from banban.routers.im import router as im_router
|
|
from banban.routers.parents import router as parents_router
|
|
from banban.routers.roles import router as roles_router
|
|
from banban.routers.wechat_auth import router as wechat_auth_router
|
|
from database.connection import get_db_manager
|
|
from initialization import init_directories
|
|
from utils.logger import session_logger
|
|
|
|
|
|
banban_router = APIRouter(prefix="/banban")
|
|
banban_router.include_router(wechat_auth_router, tags=["banban-auth"])
|
|
banban_router.include_router(bindings_router, tags=["banban-bindings"])
|
|
banban_router.include_router(children_router, tags=["banban-children"])
|
|
banban_router.include_router(devices_router, tags=["banban-devices"])
|
|
banban_router.include_router(device_im_router, tags=["banban-device-im"])
|
|
banban_router.include_router(im_router, tags=["banban-im"])
|
|
banban_router.include_router(parents_router, tags=["banban-parents"])
|
|
banban_router.include_router(roles_router, tags=["banban-roles"])
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
del app
|
|
worker_id = os.environ.get("UVICORN_WID", "0")
|
|
is_main_process = worker_id == "0"
|
|
db_manager = None
|
|
|
|
if is_main_process:
|
|
session_logger.system_info("startup", "banban standalone API starting")
|
|
|
|
init_directories()
|
|
db_manager = await get_db_manager()
|
|
|
|
if is_main_process:
|
|
session_logger.system_info("startup", "banban standalone API ready")
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
if db_manager is not None:
|
|
await db_manager.close()
|
|
if is_main_process:
|
|
session_logger.system_info("shutdown", "banban standalone API stopped")
|
|
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
configure_static_assets(app)
|
|
install_request_logging_middleware(app)
|
|
install_auth_middleware(app)
|
|
app.include_router(banban_router)
|