FastAPI：(11)SQL數據庫

由于CSDN無法展示「漸構」的「#d，#e，#t，#c，#v，#a」標簽，推薦訪問我個人網站進行閱讀：Hkini
「漸構展示」如下：在這里插入圖片描述

#c 概述文章內容概括

1.安裝SQLModel

#d SQLModel

SQLModel 是一個用于在 Python 中聲明數據庫模型的庫，它結合了 SQLAlchemy 的 ORM 能力與 Pydantic 的數據驗證特性。它旨在提供一種簡單、統一的方式來定義數據庫表結構，同時用于數據序列化、驗證和交互，主要用于與 FastAPI 等現代 Web 框架集成。
安裝SQLModel通過pip install sqlmodel

重要特征：

雙重用途（ORM + 數據驗證）：同時兼容 SQLAlchemy 的 ORM 和 Pydantic 的數據校驗與序列化。
基于類型注解：使用 Python 的類型提示定義字段和數據結構，更直觀地說明數據模型。
自動生成表結構：可通過模型類定義自動生成數據庫表。
對異步和同步均友好：兼容異步和同步數據庫操作，適應不同項目架構。
簡化模型繼承與組合：可繼承基礎類構建只讀模型、創建數據模型、更新模型等多種用途。

#e 電子病歷模型(正例) SQLModel

例子描述
醫院系統中的電子病歷模型 MedicalRecord，可表示為數據庫表，也可用于接收醫生上傳記錄或給前端返回。字段如 patient_id: int, diagnosis: str, record_time: datetime，模型用于數據庫操作和前后端數據傳輸，完全符合 SQLModel 的理念。

特征對比

? 雙重用途：模型同時作為數據庫表結構與 API 的請求/響應模型。
? 基于類型注解：所有字段通過類型注解標明類型。
? 表結構生成：可通過該模型自動創建數據表。
? 異步友好：可用于 FastAPI 異步接口處理。

from sqlmodel import Field, SQLModel, create_engine, Session
from fastapi import FastAPI
from typing import Optional
from datetime import datetimeclass MedicalRecord(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)patient_id: intdiagnosis: strrecord_time: datetimesqlite_url = "sqlite:///./test.db"
engine = create_engine(sqlite_url, echo=True)
SQLModel.metadata.create_all(engine)app = FastAPI()@app.post("/records/")
def create_record(record: MedicalRecord):with Session(engine) as session:session.add(record)session.commit()session.refresh(record)return record

#e 商品庫存模型(正例) SQLModel

例子描述
電商平臺中的 InventoryItem 模型記錄商品 ID、庫存數量、價格等信息。該模型用于庫存管理模塊的數據表定義，且用于與 API 交互中的庫存變更和查詢，統一使用。

特征對比

? 雙重用途：模型用于數據庫記錄和接口的數據通信。
? 基于類型注解：如 quantity: int, price: float。
? 自動生成表結構：可以直接映射為 inventory_items 表。
? 簡化模型繼承：可以輕松派生創建更新庫存的模型。

from sqlmodel import SQLModel, Field, create_engine, Session
from fastapi import FastAPI
from typing import Optionalclass InventoryItem(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)name: strquantity: intprice: floatengine = create_engine("sqlite:///./inventory.db", echo=True)
SQLModel.metadata.create_all(engine)app = FastAPI()@app.post("/items/")
def add_item(item: InventoryItem):with Session(engine) as session:session.add(item)session.commit()session.refresh(item)return item

#e 外部API響應模型(反例) SQLModel

例子描述
一個天氣服務中，用于解析外部 API 返回數據的 WeatherResponse 模型，字段如 temperature: float, humidity: float, location: str。該模型只用于臨時解析 JSON 響應數據，不涉及數據庫存儲，也不參與任何數據寫入操作。

特征對比

? 雙重用途缺失：僅用于解析數據，不能表示數據庫表。
? 缺乏 ORM 功能：字段雖有類型注解，但無數據庫元數據。
? 無表結構生成意義：該數據模型不參與數據庫操作。
? Pydantic 使用合理：僅作為數據驗證使用，適合用 Pydantic 而非 SQLModel。

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optionalapp = FastAPI()# 外部天氣服務返回的數據格式
class WeatherResponse(BaseModel):location: strtemperature: floathumidity: floatdescription: Optional[str]# 模擬調用外部 API，返回數據結構
@app.get("/weather/", response_model=WeatherResponse)
def get_weather():# 模擬外部 API 響應fake_api_data = {"location": "Chengdu","temperature": 28.5,"humidity": 70.2,"description": "Cloudy"}return fake_api_data

2.單一模型

#e 官網例子(正例) SQLModel

具體步驟：

創建模型
創建引擎
創建表
創建會話（Session）依賴項：Session 會存儲內存中的對象并跟蹤數據中所需更改的內容，然后它使用 engine 與數據庫進行通信。使用 yield 創建一個 FastAPI 依賴項，為每個請求提供一個新的 Session 。這確保每個請求使用一個單獨的會話。
啟動創建表：對于生產環境，可能會用一個能夠在啟動應用程序之前運行的遷移腳本如Alembic。
創建Hero類：因為每個 SQLModel 模型同時也是一個 Pydantic 模型，所以可以在與 Pydantic 模型相同的類型注釋中使用它。
讀取Hero類
讀取單個Hero
刪除單個Hero

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True): # 利用SQLModel創建數據庫模型，table=True，告訴SQLModel這是一個表模型id: int | None = Field(default=None, primary_key=True) # primary_key表示是主鍵；int | None 數據在SQL應該的是INTEGER并且NULLABLE name: str = Field(index=True) # index=True 創建SQL索引，str在數據庫中將會是TEXT或者VARCHARTage: int | None = Field(default=None, index=True)secret_name: strsqlite_file_name = "database.db" # 創建數據庫引擎，用來與數據庫保持連接
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False} # 不同線程中使用同一個SQLite數據庫，將會按照代碼結構確保「每個請求使用一個單獨的SQLModel會話」
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables(): # 創建表SQLModel.metadata.create_all(engine)def get_session(): # 創建Session會話依賴項with Session(engine) as session:yield session # 為每個請求提供一個新的SessionSessionDep = Annotated[Session, Depends(get_session)]app = FastAPI()@app.on_event("startup") # 啟動時床啊金數據庫表
def on_startup():create_db_and_tables()@app.post("/heroes/") # 創建Hero類，聲明一個Hero參數，將從json主體中讀取數據，同樣可以聲明為「返回類型」，自動生成API文檔界面
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero@app.get("/heroes/") # 讀取「Hero」類，并利用limit和offset對結果進行分頁
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}") # 讀取單個「Hero」
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.delete("/heroes/{hero_id}") # 刪除單個「Hero」
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}

3.多個模型

#c 說明多個模型

現在稍微重構一下這個應用，以提高安全性和多功能性。

如果查看之前的應用程序，可以在 UI 界面中看到，到目前為止，由客戶端決定要創建的 Hero 的 id 值。

不應該允許這樣做，因為可能會覆蓋在數據庫中已經分配的 id 。決定 id 的行為應該由后端或數據庫來完成，而非客戶端。

此外，為 hero 創建了一個 secret_name ，但到目前為止，在各處都返回了它，這就不太秘密了……😅

通過添加一些額外的模型來解決這些問題，而 SQLModel 將在這里大放異彩。

#e 官網多模型例子(正例) SQLModel

操作步驟：

創建HeroBase基類
創建Hero表模型
創建HeroPublic公共數據模型
創建Hero的數據創建模型HeroCreate
創建Hero的更新模型HeroUpdate：HeroUpdate 數據模型有些特殊，它包含創建新 hero 所需的所有相同字段，但所有字段都是可選的（都有默認值）。這樣，當更新一個 hero 時，可以只發送您想要更新的字段。因為所有字段實際上都發生了變化（類型現在包括 None ，并且現在有一個默認值 None ），需要重新聲明它們，重新聲明所有字段，因此并不是真的需要從 HeroBase 繼承。讓它繼承只是為了保持一致，但這并不必要。這更多是個人喜好的問題。
使用HeroCreate創建并返回HeroPublic：在請求中接收到一個 HeroCreate 數據模型，然后從中創建一個 Hero 表模型。這個新的表模型 Hero 會包含客戶端發送的字段，以及一個由數據庫生成的 id 。然后將與函數中相同的表模型 Hero 原樣返回。但是由于使用 HeroPublic 數據模型聲明了 response_model ，FastAPI 會使用 HeroPublic 來驗證和序列化數據。
用HeroPublic讀取Hero類
用HeroPublic讀取單個Hero
用HeroPublic更新單個Hero

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel): # 創建HeroBase基類，「共享字段」name,agename: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True): # 創建Hero表模型，繼承HeroBase的共享字段id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase): # 創建返回給API客戶端的HeroPublic模型，不包括secret_name，保護Heroid: int #　重新聲明 `id: int` 。這樣便與 API 客戶端建立了一種約定，始終可以期待 `id` 存在并且是一個整數 `int`（永遠不會是 `None` ）class HeroCreate(HeroBase):　# 創建用于創建hero的數據模型secret_name: str #不僅擁有與 `HeroBase` 相同的字段，還有 `secret_name` 。當客戶端創建一個新的hero時，會送 `secret_name` ，被存儲到數據庫中，但這些 `secret_name` 不會通過 API 返回給客戶端class HeroUpdate(HeroBase): # 創建用于更新Hero的數據模型name: str | None = Noneage: int | None = Nonesecret_name: str | None = Nonesqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()@app.post("/heroes/", response_model=HeroPublic) # 使用HeroCreate創建，并返回HeroPublic
def create_hero(hero: HeroCreate, session: SessionDep):db_hero = Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero # 將與函數中相同的表模型 `Hero` 原樣返回。但是由于使用 `HeroPublic` 數據模型聲明了 `response_model` ，**FastAPI** 會使用 `HeroPublic` 來驗證和序列化數據。@app.get("/heroes/", response_model=list[HeroPublic]) # 用HeroPublic讀取Hero表，使用 `response_model=list[HeroPublic]` 確保正確地驗證和序列化數據。
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
):heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}", response_model=HeroPublic) #用HeroPublic讀取單個Hero
def read_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.patch("/heroes/{hero_id}", response_model=HeroPublic) #用`HeroUpdate` 更新單個 Hero
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db = session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code=404, detail="Hero not found")hero_data = hero.model_dump(exclude_unset=True) #在代碼中，會得到一個 `dict` ，其中包含客戶端發送的所有數據，只有客戶端發送的數據，并排除了任何一個僅僅作為默認值存在的值。為此，用 `exclude_unset=True` 。這是最主要的技巧。hero_db.sqlmodel_update(hero_data) # 利用 `hero_data` 的數據更新 `hero_db`session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}