FastAPI 连接数据库并实现增删改查

新增一个用户：小明

用户请求接口
  ↓
FastAPI 收到请求
  ↓
SQLAlchemy 拿一张临时通行证
  ↓
去数据库查资料
  ↓
整理成 JSON
  ↓
返回给前端

大白话讲解——关系型数据库和 SQL 够用基础

fastapi-crud-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       └── users.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── user.py
│   └── schemas/
│       ├── __init__.py
│       └── user.py
├── requirements.txt
└── .env.example

mkdir fastapi-crud-demo
cd fastapi-crud-demo

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

.venv\Scripts\activate

fastapi==0.115.6
uvicorn[standard]==0.32.1
sqlalchemy==2.0.36
asyncpg==0.30.0
pydantic[email]==2.10.3
pydantic-settings==2.7.0

pip install -r requirements.txt

mkdir -p app/api/v1 app/core app/models app/schemas
touch app/__init__.py
touch app/api/__init__.py app/api/v1/__init__.py
touch app/core/__init__.py app/models/__init__.py app/schemas/__init__.py

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    APP_NAME: str = "FastAPI CRUD Demo"
    DATABASE_URL: str = "postgresql+asyncpg://postgres:password@localhost:5432/appdb"

    class Config:
        env_file = ".env"


settings = Settings()

APP_NAME=FastAPI CRUD Demo
DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/appdb

cp .env.example .env

postgresql+asyncpg://用户名:密码@主机:端口/数据库名

postgresql+asyncpg://postgres:password@localhost:5432/appdb

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)
    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):
    status: bool = True


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


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

    model_config = {"from_attributes": True}


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

model_config = {"from_attributes": True}

app/api/v1/users.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
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),
):
    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)
    )
    users = list(result.scalars().all())

    return UserListResponse(total=total, items=users)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
):
    user = (
        await db.execute(select(User).where(User.id == user_id))
    ).scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    return user


@router.post("", response_model=UserResponse)
async def create_user(
    body: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    user = User(**body.model_dump())
    db.add(user)

    try:
        await db.flush()
        await db.refresh(user)
    except IntegrityError:
        raise HTTPException(status_code=400, detail="用户名或邮箱已存在")

    return user


@router.put("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    body: UserUpdate,
    db: AsyncSession = Depends(get_db),
):
    user = (
        await db.execute(select(User).where(User.id == user_id))
    ).scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

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

    try:
        await db.flush()
        await db.refresh(user)
    except IntegrityError:
        raise HTTPException(status_code=400, detail="用户名或邮箱已存在")

    return user


@router.delete("/{user_id}")
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
):
    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": "删除成功"}

app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1 import 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(users.router, prefix="/api/v1/users", tags=["用户管理"])


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

app.include_router(...)

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1 import users
from app.core.config import settings
from app.core.database import Base, engine
from app.models import user


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


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

from app.models import user

Alembic 数据库迁移

docker run -d \
  --name fastapi-postgres \
  -e POSTGRES_DB=appdb \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=password \
  -p 5432:5432 \
  postgres:16-alpine

DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/appdb

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

http://localhost:8000/health

http://localhost:8000/docs

curl -X POST http://localhost:8000/api/v1/users \
  -H "Content-Type: application/json" \
  -d '{"username":"xiaoming","email":"xiaoming@example.com","status":true}'

{
  "username": "xiaoming",
  "email": "xiaoming@example.com",
  "id": 1,
  "status": true,
  "created_at": "2026-04-29T10:00:00Z",
  "updated_at": "2026-04-29T10:00:00Z"
}

curl "http://localhost:8000/api/v1/users?page=1&page_size=10"

{
  "total": 1,
  "items": [
    {
      "username": "xiaoming",
      "email": "xiaoming@example.com",
      "id": 1,
      "status": true,
      "created_at": "2026-04-29T10:00:00Z",
      "updated_at": "2026-04-29T10:00:00Z"
    }
  ]
}

curl http://localhost:8000/api/v1/users/1

curl -X PUT http://localhost:8000/api/v1/users/1 \
  -H "Content-Type: application/json" \
  -d '{"email":"new@example.com"}'

curl -X DELETE http://localhost:8000/api/v1/users/1

接收 page/page_size
  ↓
拼查询条件
  ↓
查 total
  ↓
查当前页数据
  ↓
返回 total + items

total = (await db.execute(count_query)).scalar() or 0
result = await db.execute(query.offset(...).limit(...))

根据 id 查询
  ↓
查不到就 404
  ↓
查到就返回

if not user:
    raise HTTPException(status_code=404, detail="用户不存在")

Schema 校验请求体
  ↓
创建 Model 对象
  ↓
db.add()
  ↓
flush()
  ↓
refresh()
  ↓
返回新对象

根据 id 查询
  ↓
查不到就 404
  ↓
只更新前端传来的字段
  ↓
flush()
  ↓
refresh()
  ↓
返回新对象

body.model_dump(exclude_unset=True)

根据 id 查询
  ↓
查不到就 404
  ↓
db.delete()
  ↓
返回删除成功

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

appdb

-e POSTGRES_DB=appdb

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

except IntegrityError:
    raise HTTPException(status_code=400, detail="用户名或邮箱已存在")

body.model_dump(exclude_unset=True)

body.model_dump()

model_config = {"from_attributes": True}

from app.models import user

alembic upgrade head

大白话讲解——FastAPI 使用 Alembic 管理数据库表结构

前端请求
  ↓
FastAPI 路由
  ↓
Pydantic 校验参数
  ↓
SQLAlchemy 操作数据库
  ↓
Pydantic 整理响应
  ↓
返回 JSON