SQLModel， 一個神奇的 Python 庫
作者：老朱2000
「SQLModel」 是一個基于 Python 類型注解的現(xiàn)代 ORM（對象關系映射）庫，由 FastAPI 的作者 Sebastián Ramírez 開發(fā)。
一直使用SQlAlchemy和Pydantic組合，有人推薦使用SQLModel。今天學習一下。
1. 什么是 SQLModel？
「SQLModel」 是一個基于 Python 類型注解的現(xiàn)代 ORM（對象關系映射）庫，由 FastAPI 的作者 Sebastián Ramírez 開發(fā)。它結合了 「Pydantic」 的數(shù)據驗證和序列化能力與 「SQLAlchemy」 的數(shù)據庫操作能力。
核心特性：
- 類型安全：完整的 Python 類型提示支持
- 代碼簡潔：使用 Python 現(xiàn)代語法，代碼量少
- 無縫集成：與 FastAPI 完美集成
- 強大功能：基于 SQLAlchemy，具備所有高級數(shù)據庫功能
- 開發(fā)體驗：優(yōu)秀的編輯器支持和自動補全
2. 使用場景
from fastapi import FastAPI
from sqlmodel import SQLModel, Field, Session, create_engine, select
app = FastAPI()
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
username: str = Field(index=True)
email: str
@app.get("/users/{user_id}")
def get_user(user_id: int, session: Session = Depends(get_session)):
user = session.get(User, user_id)
return user(2) 數(shù)據密集型應用
- 報表系統(tǒng)
- 數(shù)據分析平臺
- 內容管理系統(tǒng)
(3) 微服務架構
- 每個服務獨立的數(shù)據庫模型
- 類型安全的服務間通信
- 快速定義數(shù)據模型
- 自動生成數(shù)據庫表結構
3. 基礎示例
(1) 基本模型定義
from sqlmodel import SQLModel, Field
from typing import Optional
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = None(2) 數(shù)據庫設置
from sqlmodel import create_engine, Session
# 創(chuàng)建數(shù)據庫引擎
engine = create_engine("sqlite:///database.db")
# 創(chuàng)建表
SQLModel.metadata.create_all(engine)
# 獲取會話
def get_session():
with Session(engine) as session:
yield session(3) CRUD 操作
# 創(chuàng)建
def create_hero(name: str, secret_name: str, age: int | None = None):
with Session(engine) as session:
hero = Hero(name=name, secret_name=secret_name, age=age)
session.add(hero)
session.commit()
session.refresh(hero)
return hero
# 讀取
def get_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
# 更新
def update_hero(hero_id: int, new_age: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
hero.age = new_age
session.add(hero)
session.commit()
return hero
# 刪除
def delete_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
session.delete(hero)
session.commit()4. 進階示例
(1) 關系模型
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List
class Team(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
headquarters: str
# 一對多關系
heroes: List["Hero"] = Relationship(back_populates="team")
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = None
# 多對一關系
team_id: Optional[int] = Field(default=None, foreign_key="team.id")
team: Optional[Team] = Relationship(back_populates="heroes")(2) 復雜查詢
from sqlmodel import select, and_, or_
# 條件查詢
def get_heroes_by_age(min_age: int, max_age: int):
with Session(engine) as session:
statement = select(Hero).where(
and_(
Hero.age >= min_age,
Hero.age <= max_age
)
)
heroes = session.exec(statement).all()
return heroes
# 關聯(lián)查詢
def get_heroes_with_teams():
with Session(engine) as session:
statement = select(Hero, Team).join(Team)
results = session.exec(statement).all()
return results
# 分頁查詢
def get_heroes_paginated(skip: int = 0, limit: int = 10):
with Session(engine) as session:
statement = select(Hero).offset(skip).limit(limit)
heroes = session.exec(statement).all()
return heroes(3) 事務處理
from sqlmodel import Session
def transfer_hero_to_team(hero_id: int, new_team_id: int):
with Session(engine) as session:
try:
hero = session.get(Hero, hero_id)
new_team = session.get(Team, new_team_id)
ifnot hero ornot new_team:
raise ValueError("Hero or Team not found")
# 更新關聯(lián)
old_team_id = hero.team_id
hero.team_id = new_team_id
session.add(hero)
session.commit()
return {"message": "Transfer successful"}
except Exception as e:
session.rollback()
raise e(4) 繼承和多態(tài)
class BaseUser(SQLModel):
id: Optional[int] = Field(default=None, primary_key=True)
email: str = Field(unique=True, index=True)
is_active: bool = True
class RegularUser(BaseUser, table=True):
__tablename__ = "regular_users"
username: str
class AdminUser(BaseUser, table=True):
__tablename__ = "admin_users"
admin_level: int = Field(default=1)(5) 混合 Pydantic 模型
from pydantic import EmailStr
# 僅用于驗證的 Pydantic 模型
class UserCreate(SQLModel):
username: str
email: EmailStr
password: str
class UserResponse(SQLModel):
id: int
username: str
email: str
is_active: bool
# 數(shù)據庫模型
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
username: str = Field(index=True)
email: str = Field(unique=True, index=True)
hashed_password: str
is_active: bool = True(6) FastAPI 集成
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
app = FastAPI()
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
# 檢查用戶是否存在
existing_user = session.exec(
select(User).where(User.email == user.email)
).first()
if existing_user:
raise HTTPException(status_code=400, detail="Email already registered")
# 創(chuàng)建用戶
hashed_password = get_password_hash(user.password)
db_user = User(
username=user.username,
email=user.email,
hashed_password=hashed_password
)
session.add(db_user)
session.commit()
session.refresh(db_user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, session: Session = Depends(get_session)):
user = session.get(User, user_id)
ifnot user:
raise HTTPException(status_code=404, detail="User not found")
return user5. 優(yōu)缺點分析
(1) 優(yōu)點
① 開發(fā)效率高
# 傳統(tǒng) SQLAlchemy
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
# SQLModel - 更簡潔
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
email: str② 類型安全
# 編輯器支持自動補全和類型檢查
user = User(name="John", email="john@example.com")
print(user.name) # 編輯器知道這是字符串③ 與 FastAPI 完美集成
@app.post("/users/")
def create_user(user: UserCreate): # 自動驗證和文檔生成
return User.from_orm(user)④ 數(shù)據驗證
from pydantic import validator
class User(SQLModel, table=True):
email: str
@validator('email')
def validate_email(cls, v):
if '@' not in v:
raise ValueError('Invalid email')
return v⑤ 現(xiàn)代化語法
# 使用現(xiàn)代 Python 特性
class Config:
arbitrary_types_allowed = True(2) 缺點
① 相對較新
- 社區(qū)生態(tài)不如 Django ORM 或成熟期的 SQLAlchemy
- 可能遇到邊緣情況的 bug
② 學習曲線
# 需要理解 SQLAlchemy 和 Pydantic 的概念
# 對于新手來說概念較多③ 性能考慮
# 額外的驗證層可能帶來輕微性能開銷
# 但在大多數(shù) Web 應用中可忽略不計④ 遷移成本
- 從現(xiàn)有 SQLAlchemy 項目遷移需要重構
- 某些高級 SQLAlchemy 功能可能不支持
⑤ 文檔和資源
- 相比成熟的 ORM，學習資源較少
- 社區(qū)案例相對有限
6. 與其他 ORM 對比
(1) vs Django ORM
# Django ORM
class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField()
# SQLModel - 更 Pythonic，類型安全
class User(SQLModel, table=True):
name: str
email: str(2) vs SQLAlchemy Core
# SQLAlchemy Core - 更靈活但更冗長
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('email', String)
)
# SQLModel - 更簡潔
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
email: str(3) vs Pony ORM
# Pony ORM - 獨特的語法
class User(db.Entity):
name = Required(str)
email = Required(str)
# SQLModel - 標準的 Python 語法
class User(SQLModel, table=True):
name: str
email: str7. 優(yōu)秀實踐
(1) 項目結構
myproject/
├── models/
│ ├── __init__.py
│ ├── base.py
│ ├── user.py
│ └── post.py
├── schemas/
│ ├── request.py
│ └── response.py
├── crud/
│ └── user.py
├── database.py
└── main.py(2) 配置管理
# database.py
from sqlmodel import create_engine, Session
import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
engine = create_engine(
DATABASE_URL,
echo=True, # 開發(fā)時顯示 SQL 語句
connect_args={"check_same_thread": False} # SQLite 需要
)
def get_session():
with Session(engine) as session:
yield session(3) 錯誤處理
from sqlmodel import SQLModel, Field, Session
from contextlib import contextmanager
@contextmanager
def session_scope():
"""提供事務范圍的會話"""
session = Session(engine)
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def safe_create_user(user_data: dict):
with session_scope() as session:
user = User(**user_data)
session.add(user)
return user8. 總結
SQLModel 是一個現(xiàn)代化的 ORM 選擇，特別適合：
- 新項目：尤其是使用 FastAPI 的項目
- 類型安全要求高：需要完整類型提示的項目
- 開發(fā)效率優(yōu)先：希望減少樣板代碼的團隊
- 現(xiàn)代 Python 技術棧：使用 Python 3.6+ 新特性的項目
在選擇 SQLModel 時，需要考慮項目的具體需求、團隊的技術背景以及對新興技術的接受程度。對于大型企業(yè)級應用，可能需要評估其成熟度；而對于初創(chuàng)公司和快速迭代的項目，SQLModel 的開發(fā)效率優(yōu)勢非常明顯。
責任編輯：趙寧寧
來源：
程序員老朱
