這篇文章將教你如何在 FastAPI 專案中，使用 SQLAlchemy 來連接 SQLite 資料庫。透過這套組合，你無需安裝額外的資料庫軟體，就能以 Python 物件的方式輕鬆進行資料的新增、讀取、更新與刪除。
什麼是 SQLAlchemy 與 SQLite？
SQLite 是一個輕量級的關聯式資料庫，它不需要伺服器，整個資料庫就是一個單獨的檔案，非常適合開發與測試。SQLAlchemy 則是一個 ORM (Object-Relational Mapping，物件關聯對應) 工具，它將資料庫的資料表轉換為 Python 的類別 (Class)。
你可以把 SQLite 想像成一個 Excel 檔案，而 SQLAlchemy 是一個專屬秘書。你只需要告訴秘書「幫我新增一筆訂單」，秘書就會幫你把資料填入 Excel 的正確欄位中，你完全不需要懂複雜的 SQL 語法。資料庫基本概念
1. 引擎 (Engine)
這是 SQLAlchemy 與資料庫溝通的起點。它負責管理資料庫的連線池 (Connection Pool) 和驅動程式的呼叫。
from sqlalchemy import create_engine
# 建立 SQLite 引擎，connect_args 是 SQLite 特有的設定
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
connect_args={"check_same_thread": False} 是 SQLite 必須的設定，因為 FastAPI 是多執行緒環境，這行設定允許在不同執行緒中使用同一個連線。
2. 對話 (Session)
Session 是你與資料庫互動的暫存區。所有的查詢、新增或修改操作，都必須透過 Session 執行，最後再提交 (Commit) 到資料庫。
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# 建立 Session，之後會用它來產生獨立的資料庫連線
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
autocommit=False 確保資料不會自動寫入，讓你能在確認所有操作無誤後，再一次性提交。
3. 基底類別 (Base)
所有資料庫模型 (Model) 都必須繼承這個基底類別。它負責追蹤所有的 Model，並協助自動建立資料表。
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
之後定義的 User 或 Item 等資料表結構，都將繼承這個 Base。
資料庫實作範例
以下是一個完整的範例，包含資料庫設定、模型定義以及 API 路由。請確保你已安裝必要套件：pip install sqlalchemy。
為了避免所有程式碼擠在同一個檔案，我們將專案拆分為三個檔案：database.py (連線設定)、models.py (資料表結構)、main.py (API 路由)。
步驟 1：設定連線 (database.py)
這個檔案只負責建立引擎與 Session，是資料庫的基礎建設。
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
# 建立 SQLite 引擎
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# 建立 Session (之後用它產生連線)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 建立 ORM 模型的基底類別
Base = declarative_base()
步驟 2：定義資料表 (models.py)
這裡定義資料庫長什麼樣子。注意必須從 database.py 匯入 Base。
# models.py
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # 資料表名稱
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
步驟 3：API 路由與邏輯 (main.py)
這是程式的入口，我們在這裡組合前兩個檔案的功能。
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import select
from pydantic import BaseModel, ConfigDict
import models
from database import engine, SessionLocal
# 1. 建立資料表
# 讀取 models 中的定義，並在資料庫中建立對應的 Table
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# 2. 定義 Pydantic 模型 (用於驗證請求資料)
class UserCreate(BaseModel):
name: str
email: str
# 用於回傳給前端的資料 (包含 id)
class UserResponse(BaseModel):
id: int
name: str
email: str
# 這允許 Pydantic 模型直接從 ORM 物件（屬性存取）讀取資料，而不僅限於字典（Dict）。
model_config = ConfigDict(from_attributes=True)
# 3. 資料庫連線依賴 (Dependency)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close() # 確保請求結束後關閉連線
# 4. 建立 API 路由
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
# 使用 models.User 建立資料庫物件
db_user = models.User(name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
# 使用 models.User 進行查詢
stmt = select(models.User).where(models.User.id == user_id)
user = db.execute(stmt).scalar_one_or_none()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
get_db 配合 Depends，確保每個請求都有獨立的資料庫連線，並在結束時自動關閉，這是 FastAPI 管理資料庫連線的標準做法。
常見錯誤與解決方法
1. 忘記處理 SQLite 執行緒問題
SQLite 預設限制單一執行緒存取，但在 FastAPI 非同步環境下會導致錯誤。
# 錯誤寫法
engine = create_engine("sqlite:///./sql_app.db")
# 正確寫法
engine = create_engine(
"sqlite:///./sql_app.db",
connect_args={"check_same_thread": False} # 關鍵設定
)
2. 搞混 Pydantic 模型與 SQLAlchemy 模型
Pydantic 用於資料驗證 (API 層)，SQLAlchemy 用於資料庫操作 (DB 層)，兩者不可混用。
# 錯誤寫法 (user 是 Pydantic 模型)
db.add(user)
# 正確寫法 (需轉換為 SQLAlchemy 模型)
db_user = User(name=user.name, email=user.email)
db.add(db_user)
結語
學會 FastAPI 與 SQLAlchemy 是後端開發的基本功，重點在於正確設定 SessionLocal 與 get_db 依賴函數，這能確保資料庫連線的穩定性。
