下面將創建一個簡單的 FastAPI 應用程序，演示如何與 SQLite 數據庫進行交互。這個例子包括創建、讀取、更新和刪除（CRUD）操作。

安裝必要的包

首先，需要安裝以下包：

pip install fastapi uvicorn sqlalchemy pydantic

完整代碼示例

from fastapi import FastAPI, HTTPException, Depends, status
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import List, Optional# 創建 FastAPI 應用
app = FastAPI(title="任務管理API")# 數據庫配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.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 TaskModel(Base):__tablename__ = "tasks"id = Column(Integer, primary_key=True, index=True)title = Column(String, index=True)description = Column(String)completed = Column(Boolean, default=False)# 創建數據庫表
Base.metadata.create_all(bind=engine)# 定義 Pydantic 模型用于請求和響應
class TaskBase(BaseModel):title: strdescription: Optional[str] = Nonecompleted: bool = Falseclass TaskCreate(TaskBase):passclass Task(TaskBase):id: intclass Config:orm_mode = True# 依賴項：獲取數據庫會話
def get_db():db = SessionLocal()try:yield dbfinally:db.close()# API 路由
@app.post("/tasks/", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):db_task = TaskModel(**task.dict())db.add(db_task)db.commit()db.refresh(db_task)return db_task@app.get("/tasks/", response_model=List[Task])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):tasks = db.query(TaskModel).offset(skip).limit(limit).all()return tasks@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int, db: Session = Depends(get_db)):task = db.query(TaskModel).filter(TaskModel.id == task_id).first()if task is None:raise HTTPException(status_code=404, detail="任務未找到")return task@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task: TaskCreate, db: Session = Depends(get_db)):db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()if db_task is None:raise HTTPException(status_code=404, detail="任務未找到")for key, value in task.dict().items():setattr(db_task, key, value)db.commit()db.refresh(db_task)return db_task@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: Session = Depends(get_db)):db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()if db_task is None:raise HTTPException(status_code=404, detail="任務未找到")db.delete(db_task)db.commit()return None# 如果直接運行此文件，啟動服務器
if __name__ == "__main__":import uvicornuvicorn.run(app, host="127.0.0.1", port=8000)

運行應用

保存上面的代碼后，可以通過以下命令運行應用：

uvicorn fastapi_sql:app --reload

或者直接運行 Python 文件：

python fastapi_sql.py

使用說明

1. 啟動應用后，訪問 http://127.0.0.1:8000/docs 可以看到自動生成的 Swagger UI 文檔
2. 通過這個界面可以測試所有的 API 端點
   

API 端點說明

• POST /tasks/ - 創建新任務
• GET /tasks/ - 獲取所有任務列表
• GET /tasks/{task_id} - 獲取特定任務詳情
• PUT /tasks/{task_id} - 更新特定任務
• DELETE /tasks/{task_id} - 刪除特定任務

代碼解析

1. 數據庫設置：

   • 使用 SQLAlchemy ORM 連接 SQLite 數據庫
   • 創建數據庫模型 TaskModel 定義表結構

2. Pydantic 模型：

   • 使用 Pydantic 定義數據驗證和序列化模型
   • TaskBase 包含基本字段
   • TaskCreate 用于創建請求
   • Task 包含 ID，用于響應

3. 依賴注入：

   • 使用 Depends 注入數據庫會話
   • 確保會話在請求結束后正確關閉

4. CRUD 操作：

   • 實現了完整的增刪改查功能
   • 包含適當的錯誤處理和狀態碼

這個例子展示了 FastAPI 與 SQLAlchemy 的集成，適用于大多數數據庫交互場景。如需使用其他數據庫（如 MySQL、PostgreSQL），只需更改連接字符串和相應的驅動程序。