知栈 Logo知栈

大型工程化与 APIRouter

在大型工程中,我们需要通过 APIRouter 模块化组织路由、pydantic-settings 统一管理环境配置以及 pytest + TestClient / httpx.AsyncClient 进行自动化测试。

Mermaid 图表

APIRouter 模块化路由

1. 子模块路由定义 (app/api/v1/endpoints/users.py)

from fastapi import APIRouter
from pydantic import BaseModel, EmailStr

router = APIRouter()

class UserSchema(BaseModel):
    username: str
    email: EmailStr

@router.get("/", response_model=list[UserSchema])
async def list_users():
    return [{"username": "alice", "email": "alice@example.com"}]

2. 路由聚合分发 (app/api/v1/api.py)

from fastapi import APIRouter
from .endpoints import users

api_router = APIRouter()
api_router.include_router(users.router, prefix="/users", tags=["用户模块"])

3. 主入口挂载 (main.py)

from fastapi import FastAPI
from app.api.v1.api import api_router

app = FastAPI(title="大型工程项目")
app.include_router(api_router, prefix="/api/v1")

现代化配置管理 (pydantic-settings)

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    PROJECT_NAME: str = "FastAPI AI Platform"
    ENV: str = "development"
    DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/dbname"
    OPENAI_API_KEY: str = ""

    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

@lru_cache
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

自动化测试 (Testing with Pytest & TestClient)

FastAPI 官方推荐使用 TestClient(基于 HTTPX / Starlette)进行同步与异步接口测试:

pip install pytest pytest-asyncio httpx

编写测试用例 (tests/test_api.py)

import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport
from main import app

# 1. 经典同步测试 TestClient
def test_read_main():
    client = TestClient(app)
    response = client.get("/api/v1/users/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

# 2. 异步测试 AsyncClient
@pytest.mark.anyio
async def test_async_main():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.get("/api/v1/users/")
    assert response.status_code == 200

最后更新于