FastAPI 搭建标准的 API 服务

我要登录
我要查用户列表
我要新增订单
我要上传文件
我要查统计数据

这个接口返回 code
那个接口返回 error
这个接口不校验参数
那个接口直接拼 SQL
这个接口忘了鉴权
那个接口出错只返回 500

GET /api/v1/users?page=1&page_size=10

浏览器 / App / 小程序
  ↓
Nginx 或网关
  ↓
FastAPI 应用入口 main.py
  ↓
业务路由 users.py
  ↓
鉴权依赖 get_current_user
  ↓
数据库会话 get_db
  ↓
数据库模型 User
  ↓
响应 Schema UserListResponse
  ↓
返回 JSON

api-service/
├── app/
│   ├── __init__.py
│   ├── main.py                  # 应用入口
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── auth.py          # 登录认证接口
│   │       └── users.py         # 用户模块接口
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py            # 配置管理
│   │   ├── database.py          # 数据库连接
│   │   ├── redis.py             # Redis 连接
│   │   └── security.py          # 密码和 JWT
│   ├── middleware/
│   │   ├── __init__.py
│   │   └── auth.py              # 当前用户、管理员校验
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py              # 公共字段
│   │   └── user.py              # 用户表
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── auth.py              # 登录请求/响应格式
│   │   └── user.py              # 用户请求/响应格式
│   └── services/
│       ├── __init__.py
│       └── user_service.py      # 复杂业务逻辑，可选
├── alembic/
│   ├── env.py
│   └── versions/
├── alembic.ini
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
├── .env.example
└── README.md

mkdir api-service
cd api-service

mkdir -p app/api/v1 app/core app/middleware app/models app/schemas app/services alembic/versions
touch app/__init__.py
touch app/api/__init__.py app/api/v1/__init__.py
touch app/core/__init__.py app/middleware/__init__.py
touch app/models/__init__.py app/schemas/__init__.py app/services/__init__.py

python3 -m venv .venv
source .venv/bin/activate

.venv\Scripts\activate

fastapi==0.115.6
uvicorn[standard]==0.32.1
sqlalchemy==2.0.36
alembic==1.14.0
asyncpg==0.30.0
pydantic[email]==2.10.3
pydantic-settings==2.7.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
bcrypt==4.2.1
python-multipart==0.0.20
redis==5.2.1

pip install -r requirements.txt

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    APP_NAME: str = "API Service"
    APP_ENV: str = "development"

    DATABASE_URL: str = "postgresql+asyncpg://postgres:password@localhost:5432/appdb"
    REDIS_URL: str = "redis://localhost:6379"

    JWT_SECRET: str = "change-me-to-a-long-random-secret"
    JWT_EXPIRE_DAYS: int = 7

    class Config:
        env_file = ".env"


settings = Settings()

APP_NAME=API Service
APP_ENV=development
DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/appdb
REDIS_URL=redis://localhost:6379
JWT_SECRET=change-me-to-a-long-random-secret
JWT_EXPIRE_DAYS=7

cp .env.example .env

app/core/database.py

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings


engine = create_async_engine(settings.DATABASE_URL, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

db: AsyncSession = Depends(get_db)

app/models/base.py

import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import Mapped, mapped_column


class TimestampMixin:
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
    )

app/models/user.py

from sqlalchemy import Boolean, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
from app.models.base import TimestampMixin


class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(200), nullable=False)
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)
    status: Mapped[bool] = mapped_column(Boolean, default=True)

app/schemas/user.py

from datetime import datetime
from pydantic import BaseModel, EmailStr


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


class UserCreate(UserBase):
    password: str
    is_admin: bool = False


class UserUpdate(BaseModel):
    username: str | None = None
    email: EmailStr | None = None
    password: str | None = None
    is_admin: bool | None = None
    status: bool | None = None


class UserResponse(UserBase):
    id: int
    is_admin: bool
    status: bool
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}


class UserListResponse(BaseModel):
    total: int
    items: list[UserResponse]

app/schemas/auth.py

from pydantic import BaseModel


class LoginRequest(BaseModel):
    username: str
    password: str


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1 import auth, users
from app.core.config import settings


app = FastAPI(title=settings.APP_NAME, version="1.0.0", docs_url="/docs")

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

app.include_router(auth.router, prefix="/api/v1/auth", tags=["认证"])
app.include_router(users.router, prefix="/api/v1/users", tags=["用户管理"])


@app.get("/health")
async def health():
    return {"status": "ok"}

app/api/v1/users.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.security import hash_password
from app.middleware.auth import get_current_user, require_admin
from app.models.user import User
from app.schemas.user import UserCreate, UserListResponse, UserResponse, UserUpdate


router = APIRouter()


@router.get("", response_model=UserListResponse)
async def list_users(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    keyword: str | None = Query(None),
    db: AsyncSession = Depends(get_db),
    _current_user: User = Depends(get_current_user),
):
    query = select(User)
    count_query = select(func.count()).select_from(User)

    if keyword:
        query = query.where(User.username.ilike(f"%{keyword}%"))
        count_query = count_query.where(User.username.ilike(f"%{keyword}%"))

    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(
        query.order_by(User.id.desc()).offset((page - 1) * page_size).limit(page_size)
    )

    return UserListResponse(total=total, items=list(result.scalars().all()))


@router.post("", response_model=UserResponse)
async def create_user(
    body: UserCreate,
    db: AsyncSession = Depends(get_db),
    _current_user: User = Depends(require_admin),
):
    existing = (
        await db.execute(select(User).where(User.username == body.username))
    ).scalar_one_or_none()
    if existing:
        raise HTTPException(status_code=400, detail="用户名已存在")

    user = User(
        username=body.username,
        email=body.email,
        password_hash=hash_password(body.password),
        is_admin=body.is_admin,
    )
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user


@router.put("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    body: UserUpdate,
    db: AsyncSession = Depends(get_db),
    _current_user: User = Depends(require_admin),
):
    user = (
        await db.execute(select(User).where(User.id == user_id))
    ).scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    data = body.model_dump(exclude_unset=True)
    password = data.pop("password", None)
    if password:
        data["password_hash"] = hash_password(password)

    for key, value in data.items():
        setattr(user, key, value)

    await db.flush()
    await db.refresh(user)
    return user


@router.delete("/{user_id}")
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    _current_user: User = Depends(require_admin),
):
    user = (
        await db.execute(select(User).where(User.id == user_id))
    ).scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    await db.delete(user)
    return {"message": "删除成功"}

password = data.pop("password", None)

app/core/security.py

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def create_access_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(days=settings.JWT_EXPIRE_DAYS)
    return jwt.encode(payload, settings.JWT_SECRET, algorithm="HS256")


def decode_access_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
    except JWTError:
        return {}

app/middleware/auth.py

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.security import decode_access_token
from app.models.user import User


bearer_scheme = HTTPBearer()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = credentials.credentials
    payload = decode_access_token(token)
    user_id = payload.get("sub")

    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的 Token",
        )

    user = (
        await db.execute(select(User).where(User.id == int(user_id)))
    ).scalar_one_or_none()

    if not user or not user.status:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户不存在或已禁用",
        )

    return user


async def require_admin(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要管理员权限",
        )
    return current_user

app/api/v1/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.security import create_access_token, verify_password
from app.middleware.auth import get_current_user
from app.models.user import User
from app.schemas.auth import LoginRequest, TokenResponse
from app.schemas.user import UserResponse


router = APIRouter()


@router.post("/login", response_model=TokenResponse)
async def login(body: LoginRequest, db: AsyncSession = Depends(get_db)):
    user = (
        await db.execute(select(User).where(User.username == body.username))
    ).scalar_one_or_none()

    if not user or not verify_password(body.password, user.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )

    if not user.status:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="账号已禁用",
        )

    token = create_access_token({"sub": str(user.id)})
    return TokenResponse(access_token=token)


@router.get("/me", response_model=UserResponse)
async def me(current_user: User = Depends(get_current_user)):
    return current_user

Authorization: Bearer <access_token>

alembic init alembic

alembic/
alembic.ini

alembic/env.py

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
from app.core.config import settings
from app.core.database import Base
from app.models import user  # noqa


config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online():
    connectable = create_async_engine(settings.DATABASE_URL, poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


asyncio.run(run_migrations_online())

alembic revision --autogenerate -m "add users table"

alembic upgrade head

alembic downgrade -1

raise HTTPException(status_code=404, detail="用户不存在")

return {"error": "用户不存在"}

import logging

logger = logging.getLogger(__name__)

logger.info("user created")
logger.warning("token invalid")
logger.exception("database error")

{
  "id": 1,
  "username": "admin"
}

{
  "code": 0,
  "message": "ok",
  "data": {
    "id": 1,
    "username": "admin"
  }
}

Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"]

docker-compose.yml

services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: appdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: ${DB_PASS:-password}
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 10

  redis:
    image: redis:7-alpine
    command: redis-server --requirepass ${REDIS_PASS:-password}

  api:
    build: .
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:${DB_PASS:-password}@db:5432/appdb
      REDIS_URL: redis://:${REDIS_PASS:-password}@redis:6379
      JWT_SECRET: ${JWT_SECRET:-change-me-to-a-long-random-secret}
      APP_ENV: production
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    ports:
      - "8000:8000"

volumes:
  pgdata:

docker compose up -d --build

docker compose logs -f api

docker compose down

app/models/article.py

from sqlalchemy import Boolean, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
from app.models.base import TimestampMixin


class Article(Base, TimestampMixin):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(120), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    published: Mapped[bool] = mapped_column(Boolean, default=False)

from app.models import article  # noqa

app/schemas/article.py

from datetime import datetime
from pydantic import BaseModel


class ArticleCreate(BaseModel):
    title: str
    content: str
    published: bool = False


class ArticleUpdate(BaseModel):
    title: str | None = None
    content: str | None = None
    published: bool | None = None


class ArticleResponse(BaseModel):
    id: int
    title: str
    content: str
    published: bool
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}


class ArticleListResponse(BaseModel):
    total: int
    items: list[ArticleResponse]

app/api/v1/articles.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.middleware.auth import get_current_user, require_admin
from app.models.article import Article
from app.schemas.article import ArticleCreate, ArticleListResponse, ArticleResponse, ArticleUpdate


router = APIRouter()


@router.get("", response_model=ArticleListResponse)
async def list_articles(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    _=Depends(get_current_user),
):
    total = (await db.execute(select(func.count()).select_from(Article))).scalar() or 0
    result = await db.execute(
        select(Article).order_by(Article.id.desc()).offset((page - 1) * page_size).limit(page_size)
    )
    return ArticleListResponse(total=total, items=list(result.scalars().all()))


@router.post("", response_model=ArticleResponse)
async def create_article(
    body: ArticleCreate,
    db: AsyncSession = Depends(get_db),
    _=Depends(require_admin),
):
    article = Article(**body.model_dump())
    db.add(article)
    await db.flush()
    await db.refresh(article)
    return article


@router.put("/{article_id}", response_model=ArticleResponse)
async def update_article(
    article_id: int,
    body: ArticleUpdate,
    db: AsyncSession = Depends(get_db),
    _=Depends(require_admin),
):
    article = (
        await db.execute(select(Article).where(Article.id == article_id))
    ).scalar_one_or_none()
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")

    for key, value in body.model_dump(exclude_unset=True).items():
        setattr(article, key, value)

    await db.flush()
    await db.refresh(article)
    return article


@router.delete("/{article_id}")
async def delete_article(
    article_id: int,
    db: AsyncSession = Depends(get_db),
    _=Depends(require_admin),
):
    article = (
        await db.execute(select(Article).where(Article.id == article_id))
    ).scalar_one_or_none()
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")

    await db.delete(article)
    return {"message": "删除成功"}

from app.api.v1 import articles

app.include_router(articles.router, prefix="/api/v1/articles", tags=["文章管理"])

alembic revision --autogenerate -m "add articles table"

alembic upgrade head

uvicorn app.main:app --reload --port 8000

http://localhost:8000/docs

[ ] 1. 项目有清晰目录：api / schemas / models / core / services
[ ] 2. 配置统一从 .env 读取
[ ] 3. 数据库连接统一通过 get_db 注入
[ ] 4. 密码只存 hash，不存明文
[ ] 5. 登录成功返回 JWT
[ ] 6. 需要登录的接口使用 get_current_user
[ ] 7. 管理员接口使用 require_admin
[ ] 8. 请求体使用 Pydantic Schema 校验
[ ] 9. 响应使用 response_model 声明
[ ] 10. 列表接口有分页
[ ] 11. 业务错误使用 HTTPException
[ ] 12. 数据表变化通过 Alembic 迁移
[ ] 13. 新模型已导入 alembic/env.py
[ ] 14. /health 能正常返回
[ ] 15. /docs 能看到接口文档
[ ] 16. Docker 能正常构建和启动
[ ] 17. 生产环境 JWT_SECRET、数据库密码不是默认值

main.py 创建应用
  ↓
api/v1 注册业务路由
  ↓
schemas 校验请求和响应
  ↓
middleware 做登录鉴权
  ↓
core/database 提供数据库会话
  ↓
models 映射数据库表
  ↓
alembic 管理数据库变化
  ↓
Docker 负责部署

路径怎么命名
参数怎么校验
返回什么格式
谁能访问
数据存在哪张表
表结构怎么迁移
前端怎么联调
线上怎么部署