long_articles_agentic_supply/supply/server/app.py

"""Quart + Hypercorn 应用工厂 —— 全局配置驱动，DB + 日志 + XXL-JOB 生命周期内建"""

import logging

from quart import Quart
from quart_cors import cors

from supply.core.bootstrap import AppContext
from supply.config import GlobalConfig
from supply.infra.database.manager import DatabaseManager, create_backend
from supply.infra.observability.logging import LogService
from supply.infra.xxl_jobs import XxlJobExecutor
from supply.handlers import discover_and_register
from supply.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)

    # 初始化各层依赖 —— 全部从 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)

    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