import asyncio from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from fastapi import FastAPI from common.constant import LockConstant from common.router import auto_register_routers from config.env import AppConfig from config.get_db import close_async_engine, init_create_table from config.get_redis import RedisUtil from config.get_scheduler import SchedulerUtil from exceptions.handle import handle_exception from middlewares.handle import handle_middleware from module_admin.service.log_service import LogAggregatorService from sub_applications.handle import handle_sub_applications from utils.common_util import worship from utils.log_util import logger from utils.server_util import APIDocsUtil, IPUtil, StartupUtil from utils.transport_crypto_util import TransportKeyProvider async def _start_background_tasks(app: FastAPI) -> None: """ 启动应用后台任务 :param app: FastAPI对象 :return: None """ await SchedulerUtil.init_system_scheduler(app.state.redis) app.state.log_aggregator_task = asyncio.create_task(LogAggregatorService.consume_stream(app.state.redis)) async def _stop_background_tasks(app: FastAPI) -> None: """ 停止应用后台任务并释放资源 :param app: FastAPI对象 :return: None """ log_task = getattr(app.state, 'log_aggregator_task', None) if log_task: log_task.cancel() try: await log_task except asyncio.CancelledError: pass lock_task = getattr(app.state, 'lock_renewal_task', None) if lock_task: lock_task.cancel() try: await lock_task except asyncio.CancelledError: pass await RedisUtil.close_redis_pool(app) await SchedulerUtil.close_system_scheduler() await close_async_engine() # 生命周期事件 @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """ 应用生命周期管理 :param app: FastAPI对象 :return: None """ app.state.redis = await RedisUtil.create_redis_pool(log_enabled=False) startup_log_enabled = await StartupUtil.acquire_startup_log_gate( redis=app.state.redis, lock_key=LockConstant.APP_STARTUP_LOCK_KEY, worker_id=SchedulerUtil._worker_id, lock_expire_seconds=LockConstant.LOCK_EXPIRE_SECONDS, ) app.state.startup_log_enabled = startup_log_enabled # 获取锁成功后立即启动锁续期任务,避免初始化时间过长导致锁过期 if startup_log_enabled: app.state.lock_renewal_task = StartupUtil.start_lock_renewal( redis=app.state.redis, lock_key=LockConstant.APP_STARTUP_LOCK_KEY, worker_id=SchedulerUtil._worker_id, lock_expire_seconds=LockConstant.LOCK_EXPIRE_SECONDS, interval_seconds=LockConstant.LOCK_RENEWAL_INTERVAL, on_lock_lost=SchedulerUtil.on_lock_lost, ) with logger.contextualize(startup_phase=True, startup_log_enabled=startup_log_enabled): logger.info(f'⏰️ {AppConfig.app_name}开始启动') if startup_log_enabled: worship() TransportKeyProvider.validate_runtime_configuration() await init_create_table() await RedisUtil.check_redis_connection(app.state.redis, log_enabled=startup_log_enabled) await RedisUtil.init_sys_dict(app.state.redis) await RedisUtil.init_sys_config(app.state.redis) await _start_background_tasks(app) if startup_log_enabled: # 短暂等待确保下面的启动日志在最后打印 await asyncio.sleep(0.5) logger.info(f'🚀 {AppConfig.app_name}启动成功') host = AppConfig.app_host port = AppConfig.app_port if host == '0.0.0.0': local_ip = IPUtil.get_local_ip() network_ips = IPUtil.get_network_ips() else: local_ip = host network_ips = [host] app_links = [f'🏠 Local: http://{local_ip}:{port}'] app_links.extend(f'📡 Network: http://{ip}:{port}' for ip in network_ips) logger.opt(colors=True).info('💻 应用地址:\n' + '\n'.join(app_links)) if not AppConfig.app_disable_swagger: swagger_links = [f'🏠 Local: http://{local_ip}:{port}{APIDocsUtil.docs_url()}'] swagger_links.extend( f'📡 Network: http://{ip}:{port}{APIDocsUtil.docs_url()}' for ip in network_ips ) logger.opt(colors=True).info('📄 Swagger文档:\n' + '\n'.join(swagger_links)) if not AppConfig.app_disable_redoc: redoc_links = [f'🏠 Local: http://{local_ip}:{port}{APIDocsUtil.redoc_url()}'] redoc_links.extend( f'📡 Network: http://{ip}:{port}{APIDocsUtil.redoc_url()}' for ip in network_ips ) logger.opt(colors=True).info('📚 ReDoc文档:\n' + '\n'.join(redoc_links)) yield shutdown_log_enabled = getattr(app.state, 'startup_log_enabled', False) with logger.contextualize(startup_phase=True, startup_log_enabled=shutdown_log_enabled): await _stop_background_tasks(app) def create_app() -> FastAPI: """ 创建FastAPI应用 :return: FastAPI对象 """ # 配置API文档静态资源 APIDocsUtil.setup_docs_static_resources() # 初始化FastAPI对象 app = FastAPI( title=AppConfig.app_name, description=f'{AppConfig.app_name}接口文档', version=AppConfig.app_version, lifespan=lifespan, openapi_url=APIDocsUtil.proxy_openapi_url(), docs_url=APIDocsUtil.proxy_docs_url(), redoc_url=APIDocsUtil.proxy_redoc_url(), swagger_ui_oauth2_redirect_url=APIDocsUtil.proxy_oauth2_redirect_url(), ) # 自定义API文档路由,修复无法直接通过后端地址访问文档的问题 APIDocsUtil.custom_api_docs_router(app) # 挂载子应用 handle_sub_applications(app) # 加载中间件处理方法 handle_middleware(app) # 加载全局异常处理方法 handle_exception(app) # 自动注册路由 auto_register_routers(app) return app