爲什麼選擇FastAPI+SQLAlchemy？¶
在Python Web開發中，FastAPI以其高性能、自動生成API文檔和簡潔的語法受到廣泛歡迎；而SQLAlchemy作爲強大的ORM（對象關係映射）工具，能讓開發者用Python類操作數據庫，避免直接編寫複雜SQL語句。兩者結合可以快速構建出既高效又易於維護的數據庫驅動API，非常適合初學者入門。
準備工作：安裝依賴¶
首先，確保你的Python環境已安裝。打開終端，執行以下命令安裝所需庫：
pip install fastapi uvicorn sqlalchemy pydantic
- FastAPI：核心Web框架，用於構建API
- Uvicorn：ASGI服務器，用於運行FastAPI應用
- SQLAlchemy：ORM工具，處理數據庫交互
- Pydantic：數據驗證和序列化庫（FastAPI依賴它處理請求/響應數據）
項目結構設計¶
我們採用模塊化結構，將不同功能拆分到獨立文件中，便於維護：
myapp/
├── main.py # FastAPI主程序，定義API路由
├── database.py # 數據庫連接配置
├── models.py # SQLAlchemy ORM模型（數據庫表結構）
├── schemas.py # Pydantic模型（數據驗證/序列化）
└── crud.py # CRUD操作（增刪改查）
1. 數據庫配置（database.py）¶
這部分負責創建數據庫連接和會話管理。我們以SQLite爲例（無需額外服務器，適合初學者）：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 數據庫連接URL（SQLite無需額外配置，文件路徑即爲數據庫）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# 創建SQLAlchemy引擎（數據庫連接入口）
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} # SQLite多線程兼容
)
# 創建會話本地類（用於每次請求獲取獨立會話）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 基礎模型類（所有ORM模型需繼承此類）
Base = declarative_base()
2. 數據庫模型（models.py）¶
使用SQLAlchemy定義數據庫表結構（ORM模型），以用戶表爲例：
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) # 郵箱（唯一，索引）
age = Column(Integer, nullable=True) # 年齡（可選）
說明：
- Base 繼承自SQLAlchemy的declarative_base，是所有模型的父類
- __tablename__ 指定數據庫表名
- Column 定義表字段，需指定類型（如Integer、String）和約束（如主鍵、唯一）
3. Pydantic模型（schemas.py）¶
FastAPI用Pydantic定義數據驗證和序列化規則（區分請求和響應格式）：
from pydantic import BaseModel
# 用於創建用戶的請求數據驗證（不含ID，ID由數據庫自動生成）
class UserCreate(BaseModel):
name: str
email: str
age: int | None = None # 可選字段
# 用於返回用戶數據的響應格式（包含所有字段）
class UserResponse(BaseModel):
id: int
name: str
email: str
age: int | None = None
class Config:
orm_mode = True # 允許從ORM模型（SQLAlchemy對象）創建實例
說明：
- UserCreate 用於接收前端提交的數據（如POST請求），驗證數據合法性
- UserResponse 用於返回查詢結果，orm_mode=True 允許直接從SQLAlchemy模型轉換數據
4. CRUD操作（crud.py）¶
實現數據庫的增刪改查（Create, Read, Update, Delete）：
from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate, UserResponse
# 創建用戶
def create_user(db: Session, user: UserCreate) -> UserResponse:
db_user = User(name=user.name, email=user.email, age=user.age)
db.add(db_user)
db.commit() # 提交事務
db.refresh(db_user) # 刷新對象（獲取數據庫生成的ID）
return db_user
# 獲取單個用戶
def get_user(db: Session, user_id: int) -> UserResponse | None:
return db.query(User).filter(User.id == user_id).first()
# 獲取用戶列表
def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[UserResponse]:
return db.query(User).offset(skip).limit(limit).all()
# 更新用戶
def update_user(db: Session, user_id: int, user_update: UserCreate) -> UserResponse | None:
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return None
# 更新字段
db_user.name = user_update.name
db_user.email = user_update.email
db_user.age = user_update.age
db.commit()
db.refresh(db_user)
return db_user
# 刪除用戶
def delete_user(db: Session, user_id: int) -> bool:
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return False
db.delete(db_user)
db.commit()
return True
5. 主程序（main.py）¶
整合所有模塊，定義API路由：
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from database import engine, SessionLocal, Base
from models import User
from schemas import UserCreate, UserResponse
from crud import create_user, get_user, get_users, update_user, delete_user
# 創建數據庫表（首次運行時執行，需導入所有模型）
Base.metadata.create_all(bind=engine)
app = FastAPI(title="用戶API示例")
# 依賴項：獲取數據庫會話（每次請求自動創建/關閉會話）
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close() # 確保會話關閉，避免連接泄露
# API路由
@app.post("/users/", response_model=UserResponse)
def create_new_user(user: UserCreate, db: Session = Depends(get_db)):
"""創建新用戶"""
return create_user(db=db, user=user)
@app.get("/users/{user_id}", response_model=UserResponse | None)
def read_user(user_id: int, db: Session = Depends(get_db)):
"""獲取單個用戶"""
db_user = get_user(db=db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="用戶不存在")
return db_user
@app.get("/users/", response_model=list[UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
"""獲取用戶列表"""
users = get_users(db=db, skip=skip, limit=limit)
return users
@app.put("/users/{user_id}", response_model=UserResponse | None)
def update_existing_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
"""更新用戶信息"""
return update_user(db=db, user_id=user_id, user_update=user)
@app.delete("/users/{user_id}", response_model=bool)
def delete_existing_user(user_id: int, db: Session = Depends(get_db)):
"""刪除用戶"""
success = delete_user(db=db, user_id=user_id)
if not success:
raise HTTPException(status_code=404, detail="用戶不存在")
return success
運行與測試¶
- 啓動應用：在終端執行以下命令：
uvicorn main:app --reload
main:app表示從main.py文件導入app實例--reload開啓熱重載，代碼修改後自動重啓
- 測試API：
- 訪問Swagger UI：http://localhost:8000/docs，可交互式測試所有接口
- 或用curl命令測試：
# 創建用戶
curl -X POST "http://localhost:8000/users/" -H "Content-Type: application/json" -d '{"name":"Alice","email":"alice@example.com","age":25}'
# 獲取用戶列表
curl "http://localhost:8000/users/"
核心優勢總結¶
- FastAPI：自動生成API文檔、支持異步、數據驗證強
- SQLAlchemy：ORM簡化數據庫操作，無需手寫SQL
- 模塊化設計：分離路由、模型、驗證、業務邏輯，便於擴展
- 依賴注入：自動管理數據庫會話，避免連接泄露
通過以上步驟，你已完成一個基礎的數據庫驅動API開發。接下來可嘗試擴展功能，如添加用戶權限、實現多表關聯（如訂單表與用戶表）等。實踐是學習編程的最佳方式，動手修改代碼，探索更多FastAPI+SQLAlchemy的可能性吧！