Web Day 7 完整 CRUD:篩選、分頁與排序
執行需求:CPU 可跑。昨天的書本 API 已經能完整 CRUD,但只有基本的「列出全部」與「建立」。一旦資料量成長到幾百、幾千筆,使用者就需要更精準的控制:依作者篩選、依年份區間過濾、依標題排序、分頁載入。今天把這些進階查詢做完,並學會用 cursor-based 分頁處理大量資料的情境。學完這篇,你就具備寫一個「資料量較大但仍能快速回應」的後端所需的所有工具。
引言
真實世界的 API 跟範例教學最大的差別是「資料量」。教學用的 API 通常只有十幾筆 demo 資料,隨便取都很快;真實世界的 API 可能有幾萬、幾十萬、甚至幾億筆。這時候「把所有資料一次回傳」就會爆掉:HTTP 回應可能上百 MB、序列化成 JSON 要好幾秒、前端沒辦法一次渲染那麼多東西。因此分頁(pagination)就成了所有正式 API 的必備設計。
跟分頁配套的是篩選(filter)與排序(sort)。使用者不會想每次都看完整個資料表,他們想看的是「Hao 寫的書,依年份由新到舊,每頁 20 本」。這三個能力(篩選、排序、分頁)合在一起,就構成了所謂「Queryable API」的核心。本系列的所有真實 API 都需要這三個功能,今天把它們一次處理好。
這篇文章會做四件事:第一,設計多條件篩選(包含範圍、模糊比對、集合包含);第二,介紹兩種分頁策略(offset-based 與 cursor-based)並示範實作;第三,展示單欄位與多欄位排序;第四,把昨天的書本 API 強化成完整版本,並用 httpx 批次測試。最後會帶你用 Python 內建的 sqlite3 模組塞一萬筆假資料,體驗「真實規模」的查詢。
多條件篩選
篩選的本質是 WHERE 子句。SQLModel / SQLAlchemy 提供完整的運算子對應:
| 語意 | SQL | SQLModel 寫法 |
|---|---|---|
| 等於 | WHERE author = ? |
Book.author == "Hao" |
| 不等於 | WHERE author != ? |
Book.author != "Hao" |
| 大於 / 小於 | WHERE year > ? |
Book.year > 2020 |
| 範圍 | WHERE year BETWEEN ? AND ? |
Book.year.between(2020, 2024) |
| 包含 | WHERE author IN (?, ?) |
Book.author.in_(["Hao", "Wei"]) |
| 模糊比對 | WHERE title LIKE '%Py%' |
Book.title.contains("Py") |
| NULL | WHERE deleted_at IS NULL |
Book.deleted_at.is_(None) |
| AND | WHERE a AND b |
statement.where(a).where(b) |
| OR | WHERE a OR b |
or_(a, b) |
實務上篩選參數的設計有兩個流派:把所有可能欄位都做成選填查詢參數(簡單直覺),或把篩選條件當作一個 JSON 結構送進來(彈性高但難做規格書)。本系列用前者,原因是它跟 FastAPI 的查詢參數驗證機制整合得最好。
# 多條件篩選範例
from typing import Annotated
from fastapi import Query
from sqlmodel import Session, or_, select
@app.get("/books/filter")
def filter_books(
session: SessionDep,
author: Annotated[str | None, Query(description="作者(精確比對)")] = None,
authors: Annotated[list[str] | None, Query(description="多個作者(IN)")] = None,
year_from: Annotated[int | None, Query(ge=0)] = None,
year_to: Annotated[int | None, Query(ge=0)] = None,
keyword: Annotated[str | None, Query(min_length=1)] = None,
):
statement = select(Book)
if author is not None:
statement = statement.where(Book.author == author)
if authors is not None:
statement = statement.where(Book.author.in_(authors))
if year_from is not None:
statement = statement.where(Book.year >= year_from)
if year_to is not None:
statement = statement.where(Book.year <= year_to)
if keyword is not None:
# 在 title 或 author 裡找關鍵字(OR)
statement = statement.where(
or_(
Book.title.contains(keyword),
Book.author.contains(keyword),
)
)
return list(session.exec(statement).all())
# 範例呼叫:
# GET /books/filter?author=Hao
# GET /books/filter?authors=Hao&authors=Wei
# GET /books/filter?year_from=2020&year_to=2024
# GET /books/filter?keyword=Python
幾個寫法細節:authors: list[str] 會讓 FastAPI 自動把 ?authors=Hao&authors=Wei 解析成 ["Hao", "Wei"];or_ 必須從 sqlmodel 匯入(不是 SQLAlchemy 的 or_,兩者介面相同但模組不同)。statement.where(...).where(...) 會用 AND 連接多個條件,這是 SQLAlchemy 2.0 的慣用寫法。
分頁:offset-based 與 cursor-based
分頁有兩種主流策略,各有優缺:
| 策略 | 優點 | 缺點 | 適用情境 |
|---|---|---|---|
| offset-based(offset + limit) | 簡單直覺、可跳頁 | 大資料量時效能差、新資料插入會錯位 | 管理後台、報表 |
| cursor-based(cursor + limit) | 效能穩定、即時更新友善 | 只能「下一頁」、不能跳頁 | 無限捲動、即時動態 |
offset-based 是最直覺的設計:?offset=0&limit=20 取第 1 頁、?offset=20&limit=20 取第 2 頁。資料庫會 LIMIT 20 OFFSET 20,意思是「跳過前 20 筆、取 20 筆」。這個寫法在資料量小(例如幾千筆)時效能很好;但資料庫實際上是「先讀前 40 筆,再丟掉前 20 筆」,所以 offset 越大越慢。
cursor-based 是給大量資料用的:每次查詢都帶上一筆資料的「指標」(例如 id),下一頁查詢用 WHERE id > last_id 取之後的資料。這對資料庫來說非常友善,因為它能直接用索引定位,不需要 offset。缺點是只能「下一頁」或「上一頁」,不能直接跳到第 50 頁。
兩種分頁都用範例實作:
# 兩種分頁範例
from fastapi import Query
@app.get("/books/page")
def offset_pagination(
session: SessionDep,
offset: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
# offset-based:簡單直覺
statement = select(Book).order_by(Book.id).offset(offset).limit(limit)
items = list(session.exec(statement).all())
# 順便回傳總數(讓前端顯示「共 X 筆,目前第 Y-Z 筆」)
total = session.exec(select(func.count(Book.id))).one()
return {
"items": items,
"total": total,
"offset": offset,
"limit": limit,
}
@app.get("/books/cursor")
def cursor_pagination(
session: SessionDep,
after_id: Annotated[int | None, Query(ge=0, description="上一頁最後一筆的 id")] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
# cursor-based:高效能
statement = select(Book).order_by(Book.id).limit(limit)
if after_id is not None:
statement = statement.where(Book.id > after_id)
items = list(session.exec(statement).all())
next_cursor = items[-1].id if items else None
return {
"items": items,
"next_cursor": next_cursor,
}
# 範例呼叫:
# GET /books/page?offset=0&limit=20 -> offset 第 1 頁
# GET /books/page?offset=20&limit=20 -> offset 第 2 頁
# GET /books/cursor?limit=20 -> 第 1 頁(沒有 after_id)
# GET /books/cursor?after_id=20&limit=20 -> 從 id 20 之後開始
cursor-based 的關鍵:把「上一頁最後一筆的 id」當作下一頁的起點。這對即時更新的應用特別友善——假設你在第 1 頁讀到 id 1-20,新資料被新增並拿到 id 100-31,第 2 頁查詢 ?after_id=20 會從 id 21 開始(包含新資料),不會像 offset 一樣可能漏掉中間新增的紀錄。
排序:單欄位與多欄位
排序通常用 order_by。SQLAlchemy 2.0 提供 column.asc() 與 column.desc(),也可以直接用 + / - 表達(但可讀性差)。多欄位排序則是用逗號串接:
# 排序範例
@app.get("/books/sorted")
def sorted_books(
session: SessionDep,
sort_by: Annotated[str, Query(pattern="^(id|title|author|year)$")] = "id",
order: Annotated[str, Query(pattern="^(asc|desc)$")] = "asc",
):
column = getattr(Book, sort_by)
statement = select(Book).order_by(column.desc() if order == "desc" else column.asc())
return list(session.exec(statement).all())
@app.get("/books/multi-sorted")
def multi_sorted_books(
session: SessionDep,
# 固定寫法:先依作者,再依年份
):
statement = select(Book).order_by(Book.author.asc(), Book.year.desc())
return list(session.exec(statement).all())
注意 pattern="^(id|title|author|year)$" 的正則表達式限制了使用者能選擇的排序欄位,避免 SQL injection 或亂填欄位名稱。這個 whitelist 是寫動態排序的標準做法——不要相信使用者傳來的字串,直接組進 SQL 會被攻擊。
大量資料的處理:批次與串流
當你需要處理幾萬筆資料時,一次把所有資料讀進記憶體會爆掉。常見的解法有三種:
- 分頁(pagination):用上面兩種分頁策略,讓使用者分批取。
- 批次(batch):用
session.exec(statement).yield_per(1000)一次讀 1000 筆出來處理,處理完再讀下一批。 - 串流(streaming):用
StreamingResponse把回應切成多塊送出,前端可以一邊收一邊處理(不適合用在 JSON API 上,比較適合檔案下載)。
批次處理的範例:
# 批次讀取範例
from sqlmodel import Session, select
def export_all_books_csv() -> int:
"""把所有書寫成 CSV,回傳總筆數。示範批次處理。"""
import csv
total = 0
with Session(engine) as session:
statement = select(Book).order_by(Book.id)
# yield_per(500):一次從資料庫拿 500 筆,處理完再拿下一批
for book in session.exec(statement).yield_per(500):
# 這裡可以做任何處理(寫檔、推播、發信件...)
total += 1
return total
# 實務上會寫成 CLI 工具
# 範例輸出:共匯出 10000 本書
yield_per(500) 讓 SQLAlchemy 用 server-side cursor 一次取 500 筆,處理完才取下一批。對十萬、百萬級資料這是必備設計,否則記憶體會被吃光。本系列規模不大,但概念很重要,Day 19 背景任務會用到類似的設計。
完整實作:書本 API 強化版
把所有東西一次做完。我們建立完整的書本 API,包含多條件篩選、cursor 分頁、單/多欄位排序、用 Python 內建 sqlite3 塞一萬筆假資料,最後用 httpx 驗證查詢效能:
# src/book_api/main.py
# 書本 API 強化版:篩選 + 分頁 + 排序 + 假資料
import random
from collections.abc import Generator
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlmodel import Field as SQLField, Session, SQLModel, create_engine, func, select
DATABASE_URL = "sqlite:///./books.db"
engine = create_engine(DATABASE_URL, echo=False)
class Book(SQLModel, table=True):
id: int | None = SQLField(default=None, primary_key=True)
title: str = SQLField(min_length=1, max_length=200, index=True)
author: str = SQLField(min_length=1, max_length=80, index=True)
year: int = SQLField(ge=0, le=2100)
class BookCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
author: str = Field(min_length=1, max_length=80)
year: int = Field(ge=0, le=2100)
class BookPublic(BookCreate):
id: int
model_config = ConfigDict(from_attributes=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session() -> Generator[Session, None, None]:
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI(title="書本 API 強化版", version="0.4.0")
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.get("/books", response_model=list[BookPublic])
def list_books(
session: SessionDep,
# 篩選
author: Annotated[str | None, Query()] = None,
year_from: Annotated[int | None, Query(ge=0)] = None,
year_to: Annotated[int | None, Query(ge=0)] = None,
keyword: Annotated[str | None, Query(min_length=1)] = None,
# 排序
sort_by: Annotated[str, Query(pattern="^(id|title|author|year)$")] = "id",
order: Annotated[str, Query(pattern="^(asc|desc)$")] = "asc",
# 分頁
limit: Annotated[int, Query(ge=1, le=100)] = 20,
offset: Annotated[int, Query(ge=0)] = 0,
):
statement = select(Book)
if author is not None:
statement = statement.where(Book.author == author)
if year_from is not None:
statement = statement.where(Book.year >= year_from)
if year_to is not None:
statement = statement.where(Book.year <= year_to)
if keyword is not None:
statement = statement.where(Book.title.contains(keyword))
column = getattr(Book, sort_by)
statement = statement.order_by(column.desc() if order == "desc" else column.asc())
statement = statement.offset(offset).limit(limit)
return list(session.exec(statement).all())
@app.get("/books/cursor", response_model=list[BookPublic])
def cursor_pagination(
session: SessionDep,
after_id: Annotated[int | None, Query(ge=0)] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
statement = select(Book).order_by(Book.id).limit(limit)
if after_id is not None:
statement = statement.where(Book.id > after_id)
return list(session.exec(statement).all())
@app.post("/books", response_model=BookPublic, status_code=201)
def create_book(session: SessionDep, payload: BookCreate):
book = Book(**payload.model_dump())
session.add(book)
session.commit()
session.refresh(book)
return book
@app.get("/books/stats")
def stats(session: SessionDep):
total = session.exec(select(func.count(Book.id))).one()
avg_year = session.exec(select(func.avg(Book.year))).one()
return {"total": total, "avg_year": round(avg_year, 1) if avg_year else None}
# ---------- 管理用端點:批次塞資料(僅供測試,正式環境要拿掉) ----------
@app.post("/admin/seed")
def seed_books(session: SessionDep, count: Annotated[int, Query(ge=1, le=20000)] = 1000):
AUTHORS = ["Hao", "Wei", "Chen", "Lin", "Wang"]
for _ in range(count):
book = Book(
title=f"書名-{random.randint(0, 99999)}",
author=random.choice(AUTHORS),
year=random.randint(1990, 2025),
)
session.add(book)
session.commit()
return {"seeded": count}
這個版本把所有功能(多條件篩選、cursor 分頁、排序、批次匯入、統計)整合起來。/admin/seed 端點是測試用的,正式環境一定要拿掉(Day 13 會談到角色授權與權限隔離)。
啟動伺服器,先塞一些假資料再測試:
# tests/test_query_perf.py
import time
import httpx
BASE = "http://127.0.0.1:8000"
def main():
with httpx.Client(base_url=BASE, timeout=30.0) as client:
# 1. 塞 10000 筆假資料
t0 = time.perf_counter()
r = client.post("/admin/seed", params={"count": 10000})
t1 = time.perf_counter()
print(f"塞資料:{r.status_code} ({t1 - t0:.2f} 秒)")
# 輸出(範例):塞資料:200 (1.42 秒)
# 2. 篩選:作者是 Hao,限制 5 筆
t0 = time.perf_counter()
r = client.get("/books", params={"author": "Hao", "limit": 5})
t1 = time.perf_counter()
items = r.json()
print(f"篩選 Hao 前 5 本:{len(items)} 筆 ({t1 - t0:.3f} 秒)")
for b in items:
print(f" - id={b['id']}, author={b['author']}")
# 輸出(範例):篩選 Hao 前 5 本:5 筆 (0.012 秒)
# 3. 多條件排序(年降序、id 升序)
t0 = time.perf_counter()
r = client.get("/books", params={"sort_by": "year", "order": "desc", "limit": 3})
t1 = time.perf_counter()
items = r.json()
print(f"年降序前 3:{[(b['year'], b['id']) for b in items]} ({t1 - t0:.3f} 秒)")
# 輸出(範例):年降序前 3:[(2025, 12), (2025, 47), (2025, 89)] (0.008 秒)
# 4. Cursor 分頁:第一頁
t0 = time.perf_counter()
r = client.get("/books/cursor", params={"limit": 5})
t1 = time.perf_counter()
items = r.json()
last_id = items[-1]["id"] if items else 0
print(f"Cursor 第一頁:ids={[b['id'] for b in items]} ({t1 - t0:.3f} 秒)")
# 輸出(範例):Cursor 第一頁:ids=[1, 2, 3, 4, 5] (0.005 秒)
# 5. Cursor 分頁:第二頁
r = client.get("/books/cursor", params={"after_id": last_id, "limit": 5})
items = r.json()
print(f"Cursor 第二頁:ids={[b['id'] for b in items]}")
# 輸出(範例):Cursor 第二頁:ids=[6, 7, 8, 9, 10]
# 6. 統計
r = client.get("/books/stats")
print("統計:", r.json())
# 輸出(範例):統計: {'total': 10000, 'avg_year': 2007.4}
main()
這個迴圈做了五件事:塞一萬筆資料、篩選、多欄位排序、cursor 分頁、統計。從輸出可以看到查詢都在毫秒級完成,這是因為 SQLModel 的 SQL 編譯器把我們的 Python 表達式翻譯成 SELECT ... WHERE ... ORDER BY ... LIMIT ...,而 SQLite 在一萬筆的資料量下執行這些查詢只要幾毫秒。實際數字會依機器不同而略有差異。
常見錯誤與踩雷
第一個常見的踩雷是「用字串拼 SQL」。在 ORM 裡雖然你也能寫 session.execute(text("SELECT * FROM books")),但這會繞過 SQLAlchemy 的參數化機制,等於自己開了 SQL injection 的門。本系列一律不寫原始 SQL,全部用 ORM 表達式。如果真的有特殊需求(例如 PostgreSQL 的窗口函式),SQLAlchemy 2.0 有 with_hint 等機制安全地引入片段。
第二個是「offset 越用越慢」。一千筆內沒感覺,但一萬筆用 offset=9000 查詢會明顯變慢。這是 SQLite 與 PostgreSQL 共有的特性:offset 越大、資料庫要掃描越多無用資料。對「無限捲動」這類 UI 一定要用 cursor-based 分頁。
第三個是「忘記 ORDER BY」。Offset-based 分頁如果沒有 ORDER BY,回傳順序不固定——同一頁兩次查詢可能拿到不同結果。一定要在分頁查詢裡加上 ORDER BY,而且最好是唯一的欄位(id)。
效能與實務提醒
索引(index)是進階查詢效能的核心。我們的 Book 模型在 title 與 author 上都加了 index=True,所以「WHERE author = ?」的查詢只要 O(log N) 時間。如果之後要加「依年份查詢」,也要在 year 上加索引。索引會拖慢寫入(小幅度)、加速讀取(大幅度),原則是「對常用的 WHERE 與 ORDER BY 欄位加索引」。
當資料量繼續成長(百萬級),即使有索引 offset-based 分頁還是會變慢。這時候有三條路:換 cursor-based、加上 LIMIT 子句、把 OFFSET 的範圍限制在「可接受的最大頁數」。例如 Twitter、Facebook 的 API 都會規定 offset-based 分頁只能取前幾頁,超過就強迫用 cursor。
另外注意 .all() 會把所有查詢結果載入記憶體。如果你預期一頁可能有幾萬筆(不太可能,但實務上有遇過),改成 .yield_per(1000) 批次處理。但對「分頁」這個使用情境來說,每頁通常限制 20-100 筆,所以 .all() 是安全的。
小結
今天把完整的 CRUD 查詢做完:多條件篩選(等於、範圍、模糊、IN、OR)、兩種分頁策略(offset-based 簡單直覺,cursor-based 高效能)、單/多欄位排序、批次處理(yield_per)。我們用一萬筆假資料展示「查詢在毫秒級完成」的效能表現,並用 httpx 驗證了篩選、分頁、統計、排序的功能。本系列後續的 API 都會沿用這套結構(篩選 + 分頁 + 排序),這是「Queryable API」的標準寫法。
進階主題:查詢最佳化與 EXPLAIN
今天前面的範例在 SQLite 一萬筆下跑得很快,但真實世界的資料量與查詢模式可能讓某些查詢慢成龜速。理解 SQL 的執行計畫(execution plan)是資料庫調校的基本功,這裡示範怎麼用 Python + SQLAlchemy 看 EXPLAIN 輸出。
# explain_query.py
# 用 SQLAlchemy 看 SQL 執行計畫
import sqlite3
from sqlalchemy import text
def explain_query(db_path: str = "books.db"):
"""對某個查詢跑 EXPLAIN QUERY PLAN,看 SQLite 怎麼執行"""
conn = sqlite3.connect(db_path)
# SQLite 的 EXPLAIN QUERY PLAN 會印出每一步用到的索引或掃描方式
sql_to_explain = """
EXPLAIN QUERY PLAN
SELECT id, title FROM books WHERE author = ? ORDER BY year DESC LIMIT 10
"""
plan = conn.execute(sql_to_explain, ("Hao",)).fetchall()
for step in plan:
print(step)
conn.close()
# 範例輸出:
# (0, 0, 0, 'SEARCH books USING INDEX author_idx (author=?)')
# (0, 0, 0, 'USE TEMP B-TREE FOR ORDER BY')
def explain_with_sqlmodel(session):
"""用 SQLAlchemy 的 EXPLAIN 在正式查詢前看計畫"""
statement = text(
"EXPLAIN (ANALYZE, BUFFERS) "
"SELECT id, title FROM books WHERE author = :author"
)
plan = session.exec(statement, params={"author": "Hao"}).all()
for row in plan:
print(row)
# PostgreSQL 的 EXPLAIN ANALYZE 會印出實際執行時間與緩衝區使用
這段範例展示兩個常見的 EXPLAIN 用法。SQLite 的 EXPLAIN QUERY PLAN 簡單明瞭,會印出每一步用了哪個索引或走了全表掃描。如果看到 SCAN books(全表掃描)而不是 SEARCH books USING INDEX author_idx,就知道 author 欄位的索引沒被用上。PostgreSQL 的 EXPLAIN ANALYZE 更詳細——會印出實際執行時間、緩衝區使用量、規劃器估計,這對調校特別有用。Day 31 切到 PostgreSQL 時會正式用到。
第二個進階主題是「SQL 函式索引」。有些查詢會對欄位做函式轉換再比對,例如 WHERE LOWER(email) = ?,這種查詢無法用 email 欄位的普通索引。SQLite 與 PostgreSQL 都支援「對函式結果建立索引」,讓這種查詢也能跑得很快。
# functional_index.py
# 函式索引的設定範例
from sqlmodel import Field, Session, SQLModel, create_engine, text
class Customer(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
email: str = Field(unique=True)
# 在資料表建立後,手動加上函式索引
engine = create_engine("sqlite:///./customers.db")
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
# 對 LOWER(email) 建立索引
session.execute(text("CREATE INDEX IF NOT EXISTS idx_email_lower ON customer (LOWER(email))"))
session.commit()
# 之後 WHERE LOWER(email) = ? 就能用這個索引
函式索引特別適合「大小寫不敏感搜尋」、「前後綴去除」、「電話號碼去格式」這類業務邏輯。比在應用層先轉大小寫、再丟 SQL 查詢乾淨很多——因為查詢的結果會因為索引失效而慢,導致整個 API 變慢。
第三個進階主題是「N+1 查詢的偵測與解決」。前面提過 N+1 是 ORM 的常見效能陷阱,這裡示範怎麼在開發階段就把 N+1 抓出來:
# detect_n_plus_one.py
# 用 SQLAlchemy 的事件系統偵測 N+1
from sqlalchemy import event
from sqlalchemy.engine import Engine
@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
# 統計 SELECT 查詢的次數
if statement.strip().upper().startswith("SELECT"):
conn.info.setdefault("query_count", 0)
conn.info["query_count"] += 1
count = conn.info["query_count"]
if count > 10: # 超過 10 次 SELECT 就警告
print(f"[警告] 已執行 {count} 次 SELECT,可能有 N+1 問題")
print(f" SQL: {statement[:100]}...")
這段程式碼用 SQLAlchemy 的事件系統攔截每一次 SQL 執行,統計 SELECT 的次數。如果同一個請求裡 SELECT 次數異常高(例如取 100 筆資料卻跑了 200 次 SELECT),就警告可能有 N+1。把這段加到測試環境的設定裡,能在 CI 階段就抓到 N+1,避免上線後才發現「為什麼這個 API 突然變慢」。
實戰提醒:分頁設計的常見錯誤
分頁是 API 設計裡最容易踩雷的地方之一。這裡整理幾個我們在實戰中看過的真實錯誤,幫你提前避開。
# pagination_anti_patterns.py
# 分頁的常見錯誤與正確寫法
from sqlmodel import Session, select
# 反例 1:用 offset 但忘了 ORDER BY
def bad_pagination_no_order(session, offset, limit):
statement = select(Book).offset(offset).limit(limit)
return list(session.exec(statement).all())
# 沒 ORDER BY,回傳順序不固定;同一頁兩次查詢可能拿到不同結果
# 反例 2:用 offset 但 limit 太大
def bad_pagination_big_limit(session, offset):
statement = select(Book).offset(offset).limit(10000)
# 一頁 10000 筆根本渲染不完
return list(session.exec(statement).all())
# 反例 3:cursor 用非唯一欄位
def bad_cursor_non_unique(session, after_title):
# 用 title 當 cursor 但 title 可能重複
statement = select(Book).where(Book.title > after_title).limit(20)
return list(session.exec(statement).all())
# 如果兩本書 title 一樣,會漏資料
# 正確寫法:cursor + 唯一欄位 + 合理 limit
def good_cursor_pagination(session, after_id, limit=20):
# id 是唯一欄位、limit 限制在合理範圍
statement = select(Book).where(Book.id > after_id).order_by(Book.id).limit(limit)
return list(session.exec(statement).all())
這四個寫法示範了分頁設計的常見錯誤與正確版本。第一個錯誤「沒 ORDER BY」會讓分頁結果不穩定;第二個錯誤「limit 太大」會讓前端沒辦法渲染;第三個錯誤「非唯一欄位當 cursor」會漏資料。正確寫法是用唯一欄位(通常是 id)當 cursor,搭配合理的 limit。
另一個常見的設計陷阱是「回傳 total count 但沒做估算」。對一億筆資料的表,SELECT COUNT(*) 可能要跑幾秒鐘。很多大型 API(Twitter、GitHub)改用「是否有下一頁」的設計,不回傳 total count,這對前端體驗影響很小(前端本來就不太需要精確的總數)。
最後一個常見錯誤是「忘了處理空頁」。當資料被刪光或查詢條件沒有匹配時,你的 API 應該回傳空串列(200 + []),而不是 404 或 500。對前端來說「空串列是合法結果」,把它視為例外會讓前端邏輯複雜化。本系列所有分頁端點的回應模型都是 list[T],保證空集合也是合法回應。
結語
今天把單一資源的進階查詢做完,但目前的書本模型只有標題、作者、年份三個欄位,沒有跟「作者」這個獨立的概念連結起來。明天,我們要進入關聯設計的主題:一對多(一位作者寫多本書)、多對多(一本書有多個標籤)、一對一(書本與其詳細資料)。你會學到 SQLModel 的 Relationship 機制、外鍵(foreign key)的設計、約束與 cascade 行為,並把書本 API 升級成「作者 + 書本」雙資料表的版本。
延伸資源
- SQLModel Relationships(2025):
https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/ - SQLAlchemy 2.0 進階查詢(2024):
https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html - Use The Index, Luke:SQL 索引指南(2014,作者 Markus Winand):
https://use-the-index-luke.com/ - GitHub API Pagination(2025):
https://docs.github.com/en/rest/using-the-rest-api/using-pagination-in-the-rest-api - Designing Data-Intensive Applications(2017,Martin Kleppmann):
https://dataintensive.net/
留言
張貼留言