uv add "fastapi[standard]" sqlalchemy pymysql pydantic-settings

pip install "fastapi[standard]" sqlalchemy pymysql pydantic-settings

from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session


# SessionDep 是一個(gè)類型別名：
# 它告訴 FastAPI，只要路由參數(shù)標(biāo)注為 SessionDep，
# 就先執(zhí)行 get_db()，把得到的數(shù)據(jù)庫 Session 傳進(jìn)來。
SessionDep = Annotated[Session, Depends(get_db)]

# select(User) 表示查詢 users 表對(duì)應(yīng)的 User 模型
# where(...) 表示追加查詢條件
# scalar(...) 表示只取一條 ORM 對(duì)象結(jié)果
user = db.scalar(select(User).where(User.username == username))

mysql+pymysql://用戶名:密碼@主機(jī):端口/數(shù)據(jù)庫名?charset=utf8mb4

uv add "pymysql[rsa]"

pip install "PyMySQL[rsa]"

DATABASE_URL="mysql+pymysql://root:password@127.0.0.1:3306/todo_api?charset=utf8mb4"
DATABASE_ECHO=false
DATABASE_AUTO_CREATE=false

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # 告訴 pydantic-settings 從 .env 文件讀取配置
    # env_file_encoding 用來避免中文或特殊字符出現(xiàn)編碼問題
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # 必填配置：沒有 DATABASE_URL 時(shí)，應(yīng)用應(yīng)該直接啟動(dòng)失敗
    database_url: str

    # 是否打印 SQL。調(diào)試時(shí)可以改成 true，正常開發(fā)一般保持 false
    database_echo: bool = False

    # 是否在啟動(dòng)時(shí)自動(dòng)建表。本地 demo 可以 true，正式項(xiàng)目一般 false
    database_auto_create: bool = False


# 創(chuàng)建全局配置對(duì)象。其他模塊通過 settings.database_url 讀取配置
settings = Settings()

uv add alembic

mysql+pymysql://用戶名:密碼@主機(jī):端口/數(shù)據(jù)庫名?charset=utf8mb4

DATABASE_URL="mysql+pymysql://root:password@127.0.0.1:3306/todo_api?charset=utf8mb4"

from sqlalchemy import create_engine

from app.core.config import settings


engine = create_engine(
    # SQLAlchemy 根據(jù)這個(gè) URL 判斷數(shù)據(jù)庫類型、驅(qū)動(dòng)、賬號(hào)、密碼、主機(jī)和庫名
    settings.database_url,

    # echo=True 會(huì)把 SQL 打到控制臺(tái)，排查問題時(shí)很有用
    echo=settings.database_echo,

    # 每次從連接池取連接前先檢查連接是否還活著
    # MySQL 長時(shí)間空閑后可能主動(dòng)斷開連接，這個(gè)參數(shù)能減少斷連錯(cuò)誤
    pool_pre_ping=True,
)

from sqlalchemy.orm import sessionmaker


SessionLocal = sessionmaker(
    # 這個(gè) Session 工廠創(chuàng)建出來的 Session，都使用前面創(chuàng)建的 engine
    bind=engine,

    # 不在查詢前自動(dòng) flush，減少隱式數(shù)據(jù)庫寫入時(shí)機(jī)
    autoflush=False,

    # 不自動(dòng)提交事務(wù)。寫操作完成后必須手動(dòng) db.commit()
    autocommit=False,

    # commit 后對(duì)象字段仍然可讀，不會(huì)立刻過期
    expire_on_commit=False,
)

from sqlalchemy import select


# with 負(fù)責(zé)在代碼塊結(jié)束后關(guān)閉 Session
# 關(guān)閉 Session 不等于關(guān)閉 MySQL 服務(wù)，而是把連接還回連接池
with SessionLocal() as db:
    # select(User) 生成查詢 users 表的 SQLAlchemy 查詢對(duì)象
    # scalars(...) 表示返回 ORM 對(duì)象，而不是原始行對(duì)象
    users = db.scalars(select(User)).all()

from sqlalchemy.orm import DeclarativeBase


# 所有 ORM 模型都繼承 Base
# SQLAlchemy 會(huì)通過 Base.metadata 收集所有表結(jié)構(gòu)
class Base(DeclarativeBase):
    pass

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    # __tablename__ 指定這個(gè)模型對(duì)應(yīng)數(shù)據(jù)庫里的哪張表
    __tablename__ = "users"

    # primary_key=True 表示主鍵
    # autoincrement=True 表示由 MySQL 自動(dòng)生成遞增 ID
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # unique=True 表示數(shù)據(jù)庫層面不允許重復(fù)
    # nullable=False 表示這一列不能為空
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False)

    # default=1 是 Python 側(cè)默認(rèn)值。創(chuàng)建 User 時(shí)不傳 status，就默認(rèn)是 1
    status: Mapped[int] = mapped_column(Integer, nullable=False, default=1)

    # server_default=func.now() 表示由數(shù)據(jù)庫服務(wù)器生成默認(rèn)時(shí)間
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

from sqlalchemy import select
from sqlalchemy.orm import Session


def list_users(db: Session) -> list[User]:
    return list(db.scalars(select(User)).all())

from sqlalchemy.orm import Session


def get_user(db: Session, user_id: int) -> User | None:
    return db.get(User, user_id)

from sqlalchemy import select
from sqlalchemy.orm import Session


def get_user_by_username(db: Session, username: str) -> User | None:
    return db.scalar(select(User).where(User.username == username))

from sqlalchemy import func, select
from sqlalchemy.orm import Session


def page_users(db: Session, page: int, page_size: int) -> dict[str, object]:
    # 第 1 頁跳過 0 條，第 2 頁跳過 page_size 條
    offset = (page - 1) * page_size

    # 先查總數(shù)，用來給前端顯示一共有多少條數(shù)據(jù)
    total = db.scalar(select(func.count()).select_from(User)) or 0

    # 再查當(dāng)前頁數(shù)據(jù)
    items = db.scalars(
        select(User)
        .order_by(User.id.desc())
        .offset(offset)
        .limit(page_size)
    ).all()

    return {
        "items": list(items),
        "total": total,
        "page": page,
        "page_size": page_size,
    }

from sqlalchemy.orm import Session


def create_user(db: Session, username: str, email: str) -> User:
    # 這里只是在 Python 內(nèi)存里創(chuàng)建 ORM 對(duì)象，還沒有寫入數(shù)據(jù)庫
    user = User(
        username=username,
        email=email,
        status=1,
    )

    # 加入當(dāng)前 Session，表示這條數(shù)據(jù)準(zhǔn)備新增
    db.add(user)

    # 提交事務(wù)后，INSERT 才會(huì)真正執(zhí)行
    db.commit()

    # 刷新對(duì)象，拿到數(shù)據(jù)庫生成的 id、created_at 等字段
    db.refresh(user)

    return user

from sqlalchemy.orm import Session


def update_user_email(db: Session, user_id: int, email: str) -> User | None:
    # db.get 適合按主鍵查詢
    user = db.get(User, user_id)
    if user is None:
        return None

    # 修改 ORM 對(duì)象字段后，SQLAlchemy 會(huì)記錄這次變化
    user.email = email

    # commit 時(shí) SQLAlchemy 會(huì)生成 UPDATE 語句
    db.commit()
    db.refresh(user)

    return user

from sqlalchemy.orm import Session


def delete_user(db: Session, user_id: int) -> bool:
    user = db.get(User, user_id)
    if user is None:
        return False

    # 標(biāo)記這個(gè) ORM 對(duì)象要?jiǎng)h除
    db.delete(user)

    # commit 時(shí) SQLAlchemy 會(huì)生成 DELETE 語句
    db.commit()

    return True

from sqlalchemy.orm import Session


def create_user_with_role(db: Session, user: User, role: Role) -> User:
    try:
        # 兩個(gè)新增動(dòng)作放在同一個(gè)事務(wù)里
        db.add(user)
        db.add(role)

        # 兩個(gè)對(duì)象都寫入成功，事務(wù)才提交
        db.commit()
        db.refresh(user)
        return user
    except Exception:
        # 任意一步失敗，都撤銷本次事務(wù)里的所有數(shù)據(jù)庫變更
        db.rollback()
        raise

# begin 會(huì)自動(dòng)管理事務(wù)：
# 正常退出自動(dòng) commit，出現(xiàn)異常自動(dòng) rollback
with SessionLocal.begin() as db:
    db.add(user)
    db.add(role)

from collections.abc import Generator

from sqlalchemy.orm import Session

from app.db.session import SessionLocal


def get_db() -> Generator[Session, None, None]:
    # 每個(gè)請(qǐng)求進(jìn)來時(shí)，創(chuàng)建一個(gè)新的數(shù)據(jù)庫 Session
    db = SessionLocal()
    try:
        # yield 會(huì)把 db 交給路由函數(shù)使用
        # 路由函數(shù)執(zhí)行完成后，代碼會(huì)繼續(xù)走到 finally
        yield db
    finally:
        # 請(qǐng)求結(jié)束后關(guān)閉 Session，把連接還回連接池
        db.close()

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from app.dependencies.database import get_db
from app.modules.users.service import UserService


# prefix 表示這個(gè) router 下的接口都以 /users 開頭
# tags 用來給 OpenAPI 文檔分組
router = APIRouter(prefix="/users", tags=["users"])

# 聲明一個(gè)可復(fù)用的數(shù)據(jù)庫依賴類型
# 以后路由參數(shù)寫 db: SessionDep，就會(huì)自動(dòng)拿到 get_db 提供的 Session
SessionDep = Annotated[Session, Depends(get_db)]


@router.get("/{user_id}")
def get_user(user_id: int, db: SessionDep) -> dict[str, object]:
    # 路由層不直接寫數(shù)據(jù)庫查詢，而是調(diào)用 Service
    user = UserService.get_user(db, user_id)
    if user is None:
        # Service 返回 None，路由層把它轉(zhuǎn)換成 HTTP 404
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "id": user.id,
        "username": user.username,
        "email": user.email,
    }

app/core/config.py
  -> 讀取環(huán)境變量
app/db/session.py
  -> 創(chuàng)建 Engine、SessionLocal、init_database
app/db/base.py
  -> 定義 SQLAlchemy Base
app/db/models.py
  -> 統(tǒng)一導(dǎo)入所有 ORM 模型，方便 create_all 或 Alembic 發(fā)現(xiàn)模型
app/dependencies/database.py
  -> 定義 get_db
app/modules/users/models.py
  -> 用戶表 ORM 模型
app/modules/users/schemas.py
  -> 用戶接口的請(qǐng)求體和響應(yīng)體
app/modules/users/repository.py
  -> 用戶相關(guān)數(shù)據(jù)庫讀寫
app/modules/users/service.py
  -> 用戶業(yè)務(wù)規(guī)則
app/modules/users/router.py
  -> 用戶 HTTP 接口

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # 指定 .env 文件作為本地配置來源
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # 應(yīng)用名稱，會(huì)展示在接口文檔里
    app_name: str = "Todo API"

    # MySQL 連接串，例如 mysql+pymysql://root:password@127.0.0.1:3306/todo_api
    database_url: str

    # 是否打印 SQL
    database_echo: bool = False

    # 是否啟動(dòng)時(shí)自動(dòng) create_all 建表
    database_auto_create: bool = False


# 全局配置對(duì)象。其他模塊只讀取它，不在業(yè)務(wù)代碼里手寫環(huán)境變量解析
settings = Settings()

from sqlalchemy.orm import DeclarativeBase


# 所有 ORM 模型共同繼承的基類
# 后續(xù) User、Todo、Role 等模型都會(huì)注冊(cè)到 Base.metadata 中
class Base(DeclarativeBase):
    pass

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    # users 是數(shù)據(jù)庫里的真實(shí)表名
    __tablename__ = "users"

    # id 是主鍵，由 MySQL 自增生成
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # username 和 email 都要求唯一且不能為空
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False)

    # status 可以用來表示啟用、禁用等業(yè)務(wù)狀態(tài)
    status: Mapped[int] = mapped_column(Integer, nullable=False, default=1)

    # created_at 由數(shù)據(jù)庫寫入當(dāng)前時(shí)間
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

from app.modules.users.models import User

# 統(tǒng)一導(dǎo)出模型，方便其他地方一次性導(dǎo)入
__all__ = ["User"]

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.config import settings
from app.db.base import Base
from app.db import models


engine = create_engine(
    # 從配置讀取連接串，避免把賬號(hào)密碼寫死在代碼里
    settings.database_url,

    # 控制是否打印 SQL
    echo=settings.database_echo,

    # 防止 MySQL 空閑連接失效后，請(qǐng)求第一次查詢就報(bào)錯(cuò)
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    # SessionLocal 創(chuàng)建出的 Session 都綁定到這個(gè) engine
    bind=engine,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
)


def init_database() -> None:
    # 只在明確開啟時(shí)自動(dòng)建表
    # 正式項(xiàng)目通常使用 Alembic，不依賴 create_all
    if settings.database_auto_create:
        Base.metadata.create_all(bind=engine)

from collections.abc import Generator

from sqlalchemy.orm import Session

from app.db.session import SessionLocal


def get_db() -> Generator[Session, None, None]:
    # 為當(dāng)前請(qǐng)求創(chuàng)建 Session
    db = SessionLocal()
    try:
        # 把 Session 交給 FastAPI 路由函數(shù)
        yield db
    finally:
        # 無論接口成功還是失敗，最終都關(guān)閉 Session
        db.close()

from datetime import datetime

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    # 創(chuàng)建用戶時(shí)，前端必須傳 username 和 email
    username: str
    email: EmailStr


class UserRead(BaseModel):
    # 返回給前端的用戶字段
    id: int
    username: str
    email: EmailStr
    status: int
    created_at: datetime

uv add "pydantic[email]"

pip install "pydantic[email]"

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.modules.users.models import User


class UserRepository:
    @staticmethod
    def get_by_id(db: Session, user_id: int) -> User | None:
        # Repository 只關(guān)心數(shù)據(jù)庫怎么查
        return db.get(User, user_id)

    @staticmethod
    def get_by_username(db: Session, username: str) -> User | None:
        # 根據(jù)唯一用戶名查單個(gè)用戶
        return db.scalar(select(User).where(User.username == username))

    @staticmethod
    def create(db: Session, username: str, email: str) -> User:
        # 這里只 add，不 commit
        # commit 由 Service 在完整業(yè)務(wù)動(dòng)作結(jié)束后統(tǒng)一處理
        user = User(username=username, email=email)
        db.add(user)
        return user

from sqlalchemy.orm import Session

from app.modules.users.models import User
from app.modules.users.repository import UserRepository
from app.modules.users.schemas import UserCreate


class UserService:
    @staticmethod
    def get_user(db: Session, user_id: int) -> User | None:
        # Service 可以直接復(fù)用 Repository
        return UserRepository.get_by_id(db, user_id)

    @staticmethod
    def create_user(db: Session, data: UserCreate) -> User:
        # 業(yè)務(wù)規(guī)則：用戶名不能重復(fù)
        exists = UserRepository.get_by_username(db, data.username)
        if exists is not None:
            raise ValueError("Username already exists")

        try:
            # 先執(zhí)行數(shù)據(jù)庫新增動(dòng)作
            user = UserRepository.create(
                db,
                username=data.username,
                email=data.email,
            )

            # 業(yè)務(wù)動(dòng)作完整成功后再提交事務(wù)
            db.commit()
            db.refresh(user)
            return user
        except Exception:
            # 提交前或提交時(shí)出錯(cuò)，都回滾本次事務(wù)
            db.rollback()
            raise

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.dependencies.database import get_db
from app.modules.users.schemas import UserCreate, UserRead
from app.modules.users.service import UserService


# 這個(gè)文件只定義 users 模塊自己的 HTTP 接口
router = APIRouter(prefix="/users", tags=["users"])
SessionDep = Annotated[Session, Depends(get_db)]


@router.get("/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: SessionDep) -> UserRead:
    # 1. 路由層接收 user_id
    # 2. 把 db 和 user_id 交給 Service
    user = UserService.get_user(db, user_id)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # ORM 對(duì)象轉(zhuǎn)換成響應(yīng)模型，再由 FastAPI 轉(zhuǎn)成 JSON
    return UserRead.model_validate(user, from_attributes=True)


@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def create_user(data: UserCreate, db: SessionDep) -> UserRead:
    try:
        # data 已經(jīng)被 Pydantic 校驗(yàn)過，這里交給 Service 處理業(yè)務(wù)
        user = UserService.create_user(db, data)
    except ValueError as exc:
        # 業(yè)務(wù)異常在路由層轉(zhuǎn)換成 HTTP 狀態(tài)碼
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        ) from exc

    return UserRead.model_validate(user, from_attributes=True)

from fastapi import APIRouter

from app.modules.users.router import router as users_router


api_router = APIRouter(prefix="/api/v1")

# 把 users 模塊的接口掛到 /api/v1 下
api_router.include_router(users_router)

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.v1.router import api_router
from app.core.config import settings
from app.db.session import init_database


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # 應(yīng)用啟動(dòng)時(shí)執(zhí)行一次初始化邏輯
    init_database()

    # yield 之前是啟動(dòng)邏輯，yield 之后可以寫關(guān)閉邏輯
    yield


app = FastAPI(
    title=settings.app_name,
    # lifespan 用來管理應(yīng)用啟動(dòng)和關(guān)閉階段的邏輯
    lifespan=lifespan,
)


# 注冊(cè)總路由
app.include_router(api_router)

uv run fastapi dev app/main.py

uv run uvicorn app.main:app --reload

POST /api/v1/users
Content-Type: application/json
{
  "username": "copyer",
  "email": "copyer@example.com"
}

1. app/main.py
   FastAPI 應(yīng)用已經(jīng) include_router(api_router)
2. app/api/v1/router.py
   /api/v1 前綴命中，總路由繼續(xù)把請(qǐng)求交給 users_router
3. app/modules/users/router.py
   POST /users 命中 create_user 路由函數(shù)
4. FastAPI 發(fā)現(xiàn)路由參數(shù)里有 db: SessionDep
   于是先執(zhí)行 get_db()
5. app/dependencies/database.py
   get_db() 通過 SessionLocal() 創(chuàng)建一個(gè)新的數(shù)據(jù)庫 Session
6. router.py
   create_user(data, db) 開始執(zhí)行
   data 是 Pydantic 校驗(yàn)后的 UserCreate
   db 是 get_db() 提供的 SQLAlchemy Session
7. service.py
   UserService.create_user(db, data) 執(zhí)行業(yè)務(wù)規(guī)則
   先檢查 username 是否重復(fù)
8. repository.py
   UserRepository.get_by_username(db, data.username)
   使用 SQLAlchemy 查詢 User 表
9. SQLAlchemy
   把 select(User).where(...) 翻譯成 SELECT SQL
10. PyMySQL
    把 SELECT SQL 發(fā)給 MySQL
11. MySQL
    執(zhí)行查詢，返回是否已有這個(gè)用戶
12. service.py
    如果用戶名不存在，調(diào)用 Repository 創(chuàng)建 User 對(duì)象
13. repository.py
    db.add(user)
    把 user 標(biāo)記為準(zhǔn)備新增，但此時(shí)還沒有真正寫入數(shù)據(jù)庫
14. service.py
    db.commit()
    提交事務(wù)，SQLAlchemy 生成 INSERT SQL
15. PyMySQL
    把 INSERT SQL 發(fā)給 MySQL
16. MySQL
    寫入 users 表，生成自增 id 和默認(rèn)時(shí)間
17. service.py
    db.refresh(user)
    把數(shù)據(jù)庫生成的新字段刷新回 Python 對(duì)象
18. router.py
    把 ORM 對(duì)象轉(zhuǎn)換成 UserRead
19. FastAPI
    把 UserRead 轉(zhuǎn)成 JSON 響應(yīng)
20. get_db()
    請(qǐng)求結(jié)束，finally 執(zhí)行 db.close()
    Session 關(guān)閉，連接歸還連接池

db.add(user)

db.commit()

db.refresh(user)

db.rollback()

@router.get("/users")
def list_users():
    engine = create_engine(...)

from app.modules.users.models import User

__all__ = ["User"]