SQLAlchemy 串接關聯式資料庫

在先前的章節中，我們學會了如何接收 API 請求並用 Pydantic 驗證資料。但如果我們把伺服器關掉，這些資料就會憑空消失。要打造一個真實的商業應用，我們必須把資料永久儲存進資料庫。

雖然你可以直接寫純 SQL 語句（如 INSERT INTO users...）來操作資料庫，但在現代後端開發中，我們強烈建議使用 ORM (Object-Relational Mapping) 技術。

在 Python 界，最老牌、最穩健、也是最符合業界標準的 ORM，就是 SQLAlchemy。

1. 什麼是 ORM？為什麼要用它？

ORM 的核心概念是把資料庫中的「資料表 (Table)」映射成 Python 中的「類別 (Class)」。

你不需要寫 SELECT * FROM users，你只需要寫 db.query(User).all()。
你不需要寫 INSERT INTO，你只需要建立一個 User(name="Ken") 物件並 db.add()。

這樣做有三大好處：

防止 SQL Injection 攻擊：ORM 底層會自動過濾危險字元。
切換資料庫超容易：今天用 SQLite 測試，明天上線換成 PostgreSQL，你一行邏輯程式碼都不用改，只要換掉連線字串。
有程式碼提示：在編輯器中打 user. 會自動跳出所有欄位，手寫 SQL 是沒有這種待遇的。

2. 環境安裝與資料庫連線

首先，安裝 SQLAlchemy 以及連線 PostgreSQL 的驅動程式：

pip install sqlalchemy psycopg2-binary

接著，我們建立一個 database.py 來設定資料庫連線：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 換成你自己的資料庫網址 (格式: postgresql://帳號:密碼@網址:埠號/資料庫名)
# 這裡先示範用輕量的 SQLite 以便於本地端測試
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# 建立引擎 (SQLite 需要加 check_same_thread=False)
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# 建立 Session 工廠
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 建立所有 Model 繼承的基底類別
Base = declarative_base()

3. 定義 SQLAlchemy Models

在 models.py 中，我們定義資料表的結構。請注意，這跟上一章的 Pydantic 模型不一樣。Pydantic 負責驗證進來的 JSON，而 SQLAlchemy 負責定義存在資料庫裡的結構。

from sqlalchemy import Boolean, Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

4. 建立依賴注入 (Dependency Injection)

FastAPI 有一個非常強大的功能叫「依賴注入 (Dependencies)」。每次有人呼叫 API，我們都需要開啟一個資料庫連線 (Session)，並在 API 結束時把連線關閉還給資料庫。我們不需要在每個 API 裡面手動寫這些邏輯。

在 main.py 加上這段：

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
import models
from database import SessionLocal, engine

# 讓 SQLAlchemy 自動去建立目前還不存在的資料表
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# 定義一個 Dependency，負責提供 Database Session
def get_db():
    db = SessionLocal()
    try:
        yield db  # 將 db 交給 API 路由使用
    finally:
        db.close() # API 執行完畢後，確保連線關閉

5. 實戰：把資料寫進資料庫

現在，讓我們把 Pydantic、SQLAlchemy 和 Dependency Injection 全部結合在一起！

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

# 引入剛剛寫好的東西
from database import get_db
import models
from pydantic import BaseModel

app = FastAPI()

# 定義接收用的 Pydantic Schema
class UserCreate(BaseModel):
    email: str
    password: str

# 定義回傳用的 Pydantic Schema (隱藏密碼)
class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool

    class Config:
        from_attributes = True # 允許從 SQLAlchemy 物件直接轉換

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 1. 檢查信箱是否已註冊
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="信箱已被註冊")
    
    # 2. 建立 SQLAlchemy ORM 物件 (實務上密碼需先加密！)
    fake_hashed_password = user.password + "notreallyhashed"
    new_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    
    # 3. 寫入資料庫
    db.add(new_user)
    db.commit()
    db.refresh(new_user) # 獲取剛產生的流水號 ID
    
    # 4. 回傳給前端
    return new_user

總結

你現在已經學會了後端開發的三個基石：

路由 (Router)：接收 HTTP 請求。
驗證 (Pydantic)：確保資料格式正確。
儲存 (SQLAlchemy)：安全地與資料庫互動。

然而，在上面的範例中，密碼是明文儲存的，而且任何人都呼叫這個 API。在下一章，我們將學習資安防護的核心：密碼雜湊 (Hashing) 與 JWT 身份驗證 (Authentication)！