文章目錄
- SQLAlchemy Session對象如何操作數據庫
- SQLAlchemy非序列化對象如何返回
- 1.問題分析
- 2.解決方案
- 方法1:使用 Pydantic 響應模型(推薦)
- 方法2:手動轉換為字典(簡單快速)
- 方法3:使用 SQLAlchemy 的 as_dict 方法(需在模型中添加)
- 3.完整代碼示例
- 4.最佳實踐建議
- 路由注冊
SQLAlchemy Session對象如何操作數據庫
-
模型元數據關聯:
- 當創建模型實例時,SQLAlchemy 知道這個對象對應
__tablename__
指定的表 - 所有字段映射到表的列是通過類屬性定義的
- 當創建模型實例時,SQLAlchemy 知道這個對象對應
-
會話操作:
-
db.add(record)
:將對象加入會話的"待處理"列表 -
db.commit()
:生成并執行SQL語句:sql
INSERT INTO user_files (id, name, type, ...) VALUES (?, ?, ?, ...)
-
-
表名確定時機:
- 開發時:通過模型的
__tablename__
靜態定義 - 運行時:SQLAlchemy 通過對象的類信息確定目標表
- 開發時:通過模型的
-
為什么不需要顯式指定表名:
-
ORM 的核心特性:
- 對象關系映射自動處理表關聯
- 每個模型類隱式綁定到特定表
-
會話的通用性:
- 同一個
db
會話可操作多個模型 - 操作表取決于操作的模型對象類型
- 同一個
-
總結:
元素 | 作用 | 表名確定方式 |
---|---|---|
model 模型類 | 定義表結構 | 通過 __tablename__ 類屬性 |
record 實例 | 攜帶數據 | 實例所屬類決定目標表 |
db.add() | 加入會話 | 根據實例類型確定表 |
db.commit() | 執行操作 | 生成對應表的SQL語句 |
這種設計模式遵循SQLAlchemy的"Unit of Work"模式,開發者只需操作Python對象,ORM自動處理表映射和SQL生成。
SQLAlchemy非序列化對象如何返回
1.問題分析
-
SQLAlchemy 對象不是可序列化類型:
- 開發時創建的模型是 SQLAlchemy 模型實例
- FastAPI 的
JSONResponse
需要可序列化為 JSON 的 Python 基本類型(dict, list, str等)
-
__dict__
包含內部屬性:- 直接使用
db_user.__dict__
會包含 SQLAlchemy 內部屬性(如_sa_instance_state
) - 這些特殊屬性無法被 JSON 序列化
- 直接使用
2.解決方案
方法1:使用 Pydantic 響應模型(推薦)
from pydantic import BaseModel# 定義響應模型
class UserResponse(BaseModel):id: strusername: str@router.get("/createUser")
def create_user(db:Session = Depends(get_db)):try:# ...創建用戶的代碼...db.commit()# 使用 Pydantic 模型轉換response_data = UserResponse(id=db_user.id,username=db_user.username)return success_response(200, "創建成功", response_data.dict())except Exception as e:# ...
方法2:手動轉換為字典(簡單快速)
@router.get("/createUser")
def create_user(db: Session = Depends(get_db)):try:# ...創建用戶的代碼...db.commit()# 手動構建可序列化的字典user_data = {"id": db_user.id,"username": db_user.username}return success_response(200, "創建成功", user_data)except Exception as e:# ...
方法3:使用 SQLAlchemy 的 as_dict 方法(需在模型中添加)
# 在 User 模型類中添加
class User(Base):# ... 原有字段定義 ...def as_dict(self):return {c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs}# 在路由中使用
@router.get("/createUser")
def create_user(db: Session = Depends(get_db)):try:# ...創建用戶的代碼...db.commit()return success_response(200, "創建成功", db_user.as_dict())except Exception as e:# ...
FastAPI 內部嘗試序列化時:
- 檢測到
db_user
不是基本類型 - 嘗試調用
db_user.__dict__
- 發現
__dict__
包含不可序列化的_sa_instance_state
屬性 - 最終返回空字典
{}
作為安全處理
3.完整代碼示例
@router.get("/createUser")
def create_user(db: Session = Depends(get_db)):try:user = UserCreate(username="111")db_user = User()db_user.id = uuid.uuid1().hexdb_user.username = user.usernamedb.add(db_user)db.commit()# 解決方案:轉換為可序列化的字典user_data = {"id": db_user.id,"username": db_user.username,# 添加其他需要返回的字段}return success_response(200, "創建成功", user_data)except Exception as e:db.rollback()return error_response(500, message="用戶名至少3個字符")
4.最佳實踐建議
- 始終使用 Pydantic 模型定義響應結構
- 在數據庫操作后立即轉換為響應模型
- 避免直接返回 ORM 對象,因為它們:
- 包含數據庫會話狀態信息
- 可能導致意外數據暴露
- 難以控制序列化行為
{"code": 200,"message": "創建成功","data": {"id": "","username": ""}
}