"""Quart + Hypercorn 应用工厂 —— 全局配置驱动,DB + 日志 + XXL-JOB 生命周期内建""" import logging from quart import Quart from quart_cors import cors from src.core.bootstrap import AppContext from src.config import GlobalConfig from src.infra.database import DatabaseManager, create_backend from src.infra.observability import LogService from src.infra.xxl_jobs import XxlJobExecutor from src.config.logging_config import setup_logging from src.handlers import discover_and_register from src.handlers._utils import set_db from src.server.xxl_endpoints import set_xxl_executor logger = logging.getLogger(__name__) def create_app(global_config: GlobalConfig = None) -> Quart: """创建 Quart 应用 统一从 GlobalConfig 获取所有配置,消除散落的 Config() 调用。 可选传入 custom_config 用于测试覆盖。 Args: global_config: 可选的自定义配置,默认从环境变量加载 """ if global_config is None: global_config = GlobalConfig() logging.basicConfig(level=logging.INFO) setup_logging() # 初始化各层依赖 —— 全部从 global_config 取值 log_service = LogService(global_config.aliyun_log) db = DatabaseManager( { name: create_backend(name, config) for name, config in global_config.active_db_configs().items() } ) discover_and_register() xxl_executor = XxlJobExecutor(global_config.xxl_job) set_xxl_executor(xxl_executor) set_db(db) app = Quart(__name__) app = cors(app, allow_origin="*") # 生命周期 ctx = AppContext() ctx.on_startup(log_service.start) ctx.on_startup(db.init_pools) ctx.on_startup(xxl_executor.start) ctx.on_shutdown(xxl_executor.stop) ctx.on_shutdown(db.close_pools) ctx.on_shutdown(log_service.stop) @app.before_serving async def startup(): logger.info("Starting application...") await ctx.start_up() logger.info("Application started successfully") @app.after_serving async def shutdown(): logger.info("Shutting down application...") await ctx.shutdown() logger.info("Application shutdown successfully") return app