FastAPI+SQLite：快速搭建輕量級數據庫API服務

在現代應用開發中，快速構建數據庫驅動的API服務是常見需求。FastAPI作爲高性能Python框架，結合SQLite這一輕量級嵌入式數據庫，能幫助我們快速實現簡單但實用的後端服務。本文將通過一個完整的“學生信息管理”示例，手把手帶你從0到1搭建一個數據庫API服務。

一、環境準備：安裝必要工具¶

首先，我們需要安裝FastAPI、SQLite的Python接口（SQLAlchemy，用於ORM操作）和服務器（Uvicorn）。打開終端，執行以下命令：

pip install fastapi uvicorn sqlalchemy

FastAPI：Python高性能Web框架，自動生成API文檔，支持異步操作。
Uvicorn：ASGI服務器，用於運行FastAPI應用。
SQLAlchemy：ORM（對象關係映射）庫，讓Python代碼可直接操作數據庫，無需寫原生SQL。
SQLite：無需獨立服務器的嵌入式數據庫，適合開發和小型項目。

二、核心概念：數據模型與數據庫連接¶

我們以“學生信息”爲例，先定義數據模型和數據庫連接。創建一個main.py文件，按以下步驟編寫代碼：

1. 導入依賴¶

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

2. 定義數據模型¶

使用SQLAlchemy定義數據庫表結構（ORM模型），這裏我們創建一個Student模型，對應數據庫中的students表：

# 1. 連接SQLite數據庫（文件存儲，無需額外服務器）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 數據庫文件路徑爲當前目錄下的test.db

# 2. 創建SQLAlchemy引擎（核心連接對象）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite專用參數
)

# 3. 創建會話本地類（用於操作數據庫）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 4. 定義基類（所有ORM模型繼承自它）
Base = declarative_base()

# 5. 定義Student模型（對應數據庫表）
class Student(Base):
    __tablename__ = "students"  # 表名

    id = Column(Integer, primary_key=True, index=True)  # 主鍵，自增
    name = Column(String, index=True, nullable=False)   # 學生姓名（非空）
    age = Column(Integer, nullable=True)               # 學生年齡（可選）

3. 創建數據庫表¶

在第一次運行時，需要創建表結構。可以通過SQLAlchemy的create_all()方法自動生成表：

# 創建所有表（需在定義完模型後執行）
Base.metadata.create_all(bind=engine)

4. 定義Pydantic模型（數據驗證與序列化）¶

FastAPI需要明確請求/響應的數據格式，我們用Pydantic定義數據模型：

# 用於請求數據的模型（無需id，由數據庫自動生成）
class StudentCreate(BaseModel):
    name: str
    age: int | None = None  # 年齡可選

# 用於響應數據的模型（包含id）
class StudentResponse(BaseModel):
    id: int
    name: str
    age: int | None = None

    class Config:
        orm_mode = True  # 允許從ORM對象（SQLAlchemy模型）中讀取數據

三、數據庫操作函數（CRUD）¶

接下來，編寫操作數據庫的函數（CRUD：創建、讀取、更新、刪除）。我們通過SQLAlchemy的會話（Session）執行數據庫操作：

# 依賴項：獲取數據庫會話（每次請求生成新會話，用完關閉）
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 1. 創建學生（POST /students）
def create_student(db: Session, student: StudentCreate):
    db_student = Student(name=student.name, age=student.age)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)  # 刷新數據，獲取數據庫生成的id
    return db_student

# 2. 獲取單個學生（GET /students/{student_id}）
def get_student(db: Session, student_id: int):
    return db.query(Student).filter(Student.id == student_id).first()

# 3. 獲取所有學生（GET /students）
def get_all_students(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Student).offset(skip).limit(limit).all()

# 4. 更新學生（PUT /students/{student_id}）
def update_student(db: Session, student_id: int, student_update: StudentCreate):
    db_student = get_student(db, student_id)
    if not db_student:
        return None
    # 更新字段（僅name和age，id不可修改）
    db_student.name = student_update.name
    db_student.age = student_update.age
    db.commit()
    db.refresh(db_student)
    return db_student

# 5. 刪除學生（DELETE /students/{student_id}）
def delete_student(db: Session, student_id: int):
    db_student = get_student(db, student_id)
    if not db_student:
        return False
    db.delete(db_student)
    db.commit()
    return True

四、搭建FastAPI路由（API端點）¶

現在，將CRUD函數與FastAPI路由綁定，實現具體的API接口：

# 創建FastAPI實例
app = FastAPI(title="學生信息管理API")

# 1. 創建學生（POST請求）
@app.post("/students", response_model=StudentResponse, status_code=201)
def create_student_endpoint(student: StudentCreate, db: Session = Depends(get_db)):
    db_student = create_student(db=db, student=student)
    return db_student

# 2. 獲取單個學生（GET請求）
@app.get("/students/{student_id}", response_model=StudentResponse)
def get_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    db_student = get_student(db, student_id)
    if not db_student:
        raise HTTPException(status_code=404, detail="學生不存在")
    return db_student

# 3. 獲取所有學生（GET請求）
@app.get("/students", response_model=list[StudentResponse])
def get_all_students_endpoint(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    students = get_all_students(db, skip=skip, limit=limit)
    return students

# 4. 更新學生（PUT請求）
@app.put("/students/{student_id}", response_model=StudentResponse)
def update_student_endpoint(student_id: int, student: StudentCreate, db: Session = Depends(get_db)):
    db_student = update_student(db, student_id, student)
    if not db_student:
        raise HTTPException(status_code=404, detail="學生不存在")
    return db_student

# 5. 刪除學生（DELETE請求）
@app.delete("/students/{student_id}", status_code=204)
def delete_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    success = delete_student(db, student_id)
    if not success:
        raise HTTPException(status_code=404, detail="學生不存在")
    return None

五、運行與測試API服務¶

1. 啓動應用¶

在終端執行以下命令啓動服務（--reload開啓熱重載，修改代碼後自動重啓）：

uvicorn main:app --reload

main：文件名（main.py）
app：FastAPI實例名
啓動後，訪問 http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc 可查看交互式API文檔。

2. 測試API¶

使用FastAPI的交互式文檔（Swagger UI）或命令行工具測試：

創建學生：
點擊 /docs 中的 POST /students → 點擊“Try it out” → 輸入JSON數據（如 {"name": "Alice", "age": 20}）→ 點擊“Execute”。
獲取單個學生：
點擊 GET /students/{student_id} → 輸入剛創建的學生ID（如 1）→ 點擊“Execute”。
其他操作：
同理測試 GET /students（獲取所有）、PUT /students/{id}（更新）、DELETE /students/{id}（刪除）。

六、總結¶

通過本文，你已完成一個基於FastAPI+SQLite的輕量級數據庫API服務。核心優勢包括：

輕量易用：SQLite無需額外服務器，FastAPI自動生成API文檔，降低開發門檻。
高性能：FastAPI支持異步操作，適合高併發場景（但SQLite本身是單線程，適合中小規模）。
擴展性強：可通過修改SQLAlchemy模型和路由，快速支持複雜業務（如多表關聯、權限控制）。

後續可嘗試擴展：添加分頁、過濾、排序功能，或遷移到PostgreSQL/MySQL等數據庫。

完整代碼示例（可直接複製到main.py運行）：

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# 數據庫配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 學生模型（數據庫表）
class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    age = Column(Integer, nullable=True)

Base.metadata.create_all(bind=engine)  # 創建表

# Pydantic模型（數據驗證）
class StudentCreate(BaseModel):
    name: str
    age: int | None = None

class StudentResponse(BaseModel):
    id: int
    name: str
    age: int | None = None
    class Config:
        orm_mode = True

# 數據庫會話依賴
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 數據庫操作函數
def create_student(db: Session, student: StudentCreate):
    db_student = Student(name=student.name, age=student.age)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

def get_student(db: Session, student_id: int):
    return db.query(Student).filter(Student.id == student_id).first()

def get_all_students(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Student).offset(skip).limit(limit).all()

def update_student(db: Session, student_id: int, student: StudentCreate):
    db_student = get_student(db, student_id)
    if not db_student:
        return None
    db_student.name = student.name
    db_student.age = student.age
    db.commit()
    db.refresh(db_student)
    return db_student

def delete_student(db: Session, student_id: int):
    db_student = get_student(db, student_id)
    if not db_student:
        return False
    db.delete(db_student)
    db.commit()
    return True

# FastAPI應用
app = FastAPI(title="學生信息API")

# API路由
@app.post("/students", response_model=StudentResponse, status_code=201)
def create_student_endpoint(student: StudentCreate, db: Session = Depends(get_db)):
    return create_student(db, student)

@app.get("/students/{student_id}", response_model=StudentResponse)
def get_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    student = get_student(db, student_id)
    if not student:
        raise HTTPException(status_code=404, detail="學生不存在")
    return student

@app.get("/students", response_model=list[StudentResponse])
def get_all_students_endpoint(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return get_all_students(db, skip, limit)

@app.put("/students/{student_id}", response_model=StudentResponse)
def update_student_endpoint(student_id: int, student: StudentCreate, db: Session = Depends(get_db)):
    updated = update_student(db, student_id, student)
    if not updated:
        raise HTTPException(status_code=404, detail="學生不存在")
    return updated

@app.delete("/students/{student_id}", status_code=204)
def delete_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    success = delete_student(db, student_id)
    if not success:
        raise HTTPException(status_code=404, detail="學生不存在")
    return None