test/app/main.py

"""应用入口。

启动方式：
    uvicorn app.main:app --reload          # 开发
    python -m app.main                     # 等价写法（会先自动建表）

接口文档：
    http://127.0.0.1:8000/docs             # Swagger UI
    http://127.0.0.1:8000/redoc            # ReDoc
    http://127.0.0.1:8000/                 # 前端页面
"""

from __future__ import annotations

import logging
import sys
import time
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

# --- 让「直接运行本文件」也能工作 -------------------------------------------
# `python app/main.py`（或 PyCharm 里右键 Run）时，Python 把 sys.path[0]
# 设成**脚本所在目录** app/，于是下面那行 `from app.api import ...`
# 找不到 app 包，报 ModuleNotFoundError: No module named 'app'。
# 把项目根补进 sys.path，这条路才通。
# 用 `python -m app.main` / `uvicorn app.main:app` / 项目根的 run.py 启动时，
# 项目根本来就在 sys.path 里，这几行是无害的。
_ROOT = str(Path(__file__).resolve().parent.parent)
if _ROOT not in sys.path:
    sys.path.insert(0, _ROOT)
# ---------------------------------------------------------------------------

from app.api import api_router
from app.core.config import settings
from app.core.database import engine, ensure_database_exists
from app.core.exceptions import BusinessError
from app.model import Base

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-7s  %(name)s  %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger("wolin")

STATIC_DIR = Path(__file__).parent / "static"


@asynccontextmanager
async def lifespan(app: FastAPI):
    ensure_database_exists()
    Base.metadata.create_all(bind=engine)
    logger.info("数据库就绪：%s", settings.DB_NAME)
    logger.info("接口文档：http://127.0.0.1:8000/docs")
    yield
    engine.dispose()


app = FastAPI(
    title=settings.APP_NAME,
    description=(
        f"{settings.APP_DESC}\n\n"
        "**统一响应格式**：`{code, msg, data}`，`code=0` 表示成功。\n\n"
        "**鉴权**：除 `/auth/login` 外都需要在请求头带 `Authorization: Bearer <token>`。\n\n"
        "**分层**：api（路由）/ service（业务）/ dao（数据访问）/ model（ORM）/ schema（校验）。"
    ),
    version=settings.VERSION,
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ==================================================================== 异常处理
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(status_code=200, content={"code": exc.code, "msg": exc.msg, "data": exc.data})


@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    """把 Pydantic 的报错翻译成人话，前端可以直接弹。"""
    problems = []
    for error in exc.errors():
        location = " → ".join(str(x) for x in error.get("loc", []) if x != "body")
        problems.append(f"{location}: {error.get('msg')}")
    return JSONResponse(
        status_code=200,
        content={"code": 422, "msg": "参数校验未通过：" + "；".join(problems), "data": {"errors": problems}},
    )


@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    logger.warning("数据库约束冲突：%s", exc)
    return JSONResponse(status_code=200, content={"code": 409, "msg": "数据冲突（唯一键重复或外键不合法）", "data": None})


@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
    logger.exception("数据库错误")
    return JSONResponse(status_code=200, content={"code": 500, "msg": f"数据库操作失败：{exc}", "data": None})


@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception):
    logger.exception("未处理异常：%s %s", request.method, request.url.path)
    return JSONResponse(status_code=200, content={"code": 500, "msg": f"服务器内部错误：{exc}", "data": None})


# ==================================================================== 中间件
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Process-Time-ms"] = f"{(time.perf_counter() - start) * 1000:.1f}"
    return response


# ==================================================================== 路由
app.include_router(api_router)


@app.get("/api/health", tags=["系统"], summary="健康检查")
def health():
    return {
        "code": 0,
        "msg": "ok",
        "data": {
            "app": settings.APP_NAME,
            "version": settings.VERSION,
            "database": settings.DB_NAME,
            "dialect": engine.dialect.name,
            "auth_enabled": settings.AUTH_ENABLED,
        },
    }


# ==================================================================== 前端
if STATIC_DIR.exists():
    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

    @app.get("/", include_in_schema=False)
    def index():
        return FileResponse(str(STATIC_DIR / "index.html"))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=False)