Files
2026-09-21 11:19:07 +08:00

87 lines
3.3 KiB
Python

"""FastAPI 应用工厂。"""
from fastapi import FastAPI
from app.auth.routes import router as auth_router
from app.avatars.routes import router as avatars_router
from app.characters.routes import router as characters_router
from app.community.routes import router as community_router
from app.conversations.routes import message_router, run_router
from app.conversations.routes import router as conversations_router
from app.core.config import get_settings
from app.core.errors import install_error_handlers
from app.core.middleware import RequestContextMiddleware, configure_logging
from app.db.session import init_engine
from app.marketplace.routes import finance_router, webhook_router
from app.marketplace.routes import router as marketplace_router
from app.media.routes import router as media_router
from app.memories.routes import cycle_router
from app.memories.routes import router as memories_router
from app.ops.routes import admin_router, ops_router
from app.plugins.routes import inst_router, invoke_router
from app.plugins.routes import router as plugins_router
from app.tools.routes import calls_router, tools_router
from app.workspaces.routes import router as workspaces_router
def create_app() -> FastAPI:
settings = get_settings()
configure_logging(settings.log_level)
app = FastAPI(
title="Hunjing API",
version="0.1.0",
# 公共契约位于 docs/contracts 与 openapi/;本服务 docs 仅供开发调试。
docs_url="/_dev/docs",
openapi_url="/_dev/openapi.json",
)
install_error_handlers(app)
init_engine(app)
# 开发档 CORS:SSE 等流式响应不经前端代理直连本服务(代理会缓冲流)。
# 生产由反向代理同源承担,不依赖此配置。
if settings.env == "development":
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000", "http://127.0.0.1:3000",
"http://localhost:3001", "http://127.0.0.1:3001",
"http://localhost:3002", "http://127.0.0.1:3002",
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
)
app.add_middleware(RequestContextMiddleware)
app.include_router(auth_router)
app.include_router(conversations_router)
app.include_router(run_router)
app.include_router(message_router)
app.include_router(cycle_router) # 先于 characters:避免通配生命周期路由拦截
app.include_router(characters_router)
app.include_router(memories_router)
app.include_router(tools_router)
app.include_router(calls_router)
app.include_router(marketplace_router)
app.include_router(finance_router)
app.include_router(webhook_router)
app.include_router(community_router)
app.include_router(workspaces_router)
app.include_router(plugins_router)
app.include_router(inst_router)
app.include_router(invoke_router)
app.include_router(media_router)
app.include_router(avatars_router)
app.include_router(ops_router)
app.include_router(admin_router)
@app.get("/healthz", tags=["ops"])
async def healthz() -> dict[str, str]:
return {"status": "ok", "env": settings.env}
return app
app = create_app()