YG_FT/backend/app/main.py

import asyncio
from contextlib import suppress

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.cache_paths import setup_local_caches

# 在任何 docling / tiktoken 模块被实例化之前设置缓存路径，避免首调用走到 ~/.cache。
setup_local_caches()

from app.api.v1.router import api_router  # noqa: E402
from app.core.config import docs_kwargs, get_settings  # noqa: E402
from app.core.logging import configure_logging, setup_request_logging  # noqa: E402
from app.workers.compute_poller import run_compute_poller  # noqa: E402


def create_app() -> FastAPI:
    settings = get_settings()
    configure_logging(settings)

    app = FastAPI(title=settings.app_name, **docs_kwargs(settings.enable_docs))
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_allow_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    setup_request_logging(app)
    app.include_router(api_router, prefix=settings.route_prefix)

    @app.on_event("startup")
    async def start_workers() -> None:
        app.state.compute_poller_task = asyncio.create_task(run_compute_poller())

    @app.on_event("shutdown")
    async def stop_workers() -> None:
        task = getattr(app.state, "compute_poller_task", None)
        if task:
            task.cancel()
            with suppress(asyncio.CancelledError):
                await task

    return app


app = create_app()