概述

SQLModel 是一个 ORM 框架，其基于 SQLAlchemy 和 Pydantic，其中 SQLALchemy 提供底层 ORM 能力，Pydantic 提供类型校验能力，SQLModel 中，一个 SQLModel model 既是一个 SQLAlchemy model 也是一个 Pydantic model。

SQLModel 的优势在于解决了 Python Web 开发中最大的痛点之一，ORM model 与 view model 重复问题。

通常，无论任何语言，Web 开发会将代码分层，常见的是分为三层：

1. controller 层：将 view model 返回给前端，view model 是要返回的数据，会在这个层对入参或者出参进行参数校验，参数校验通常会定义一个 view model，view model 中定义各种参数限制
2. service 层：是业务在代码层面的具体实现
3. dao 层：负责与数据库交互，与具体业务无关，会使用 ORM 框架，定义一个一个 ORM model

从数据库查询数据，查询出来是一个 ORM model 对象，操作这个 ORM model 对象就相当于操作数据库，然而要返回的数据不会是一个单纯的 ORM model，通常是 ORM model 中的数据以及其他数据共同组成的一团数据，通常，会将 ORM nmodel 转为 dict 而后整体合并数据。等到要返回数据时，再将 dict 数据转为 view model 数据，最后由框架处理 view model 并将其返回（通常转为json后返回）。

上面的问题在于，controller 层与 dao 层都需要一个model，service 还要做合并，导致每两层之间必须都要做一次转换。

SQLModel 解决了这一问题，SQLModel 中，一个 SQLModel model 既是一个 SQLAlchemy model 也是一个 Pydantic model，由此将 controller 层 view model 和 dao 层 ORM model 统一，由此 service 层也可考虑使用 Pydantic model，或者还是使用 dict，Pydantic 原生支持转 dict。

快速开始

pip install sqlmodel

样例

from typing import Optional

from sqlmodel import Field, Session, SQLModel, create_engine


# 对应数据库中一张表。继承 SQLModel，写明 table=True
class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


# create_engine，是间接引用 SQLAlchemy 中的 create_engine 函数
engine = create_engine("mysql+pymysql://root:mYsql123456_@127.0.0.1:3306/dev")

# 创建表
SQLModel.metadata.create_all(engine)

# 创建 model 对象，既是 SQLAlchemy model 对象，也是 Pydantic 对象
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

# 插入数据，Session 是 sqlalchemy.orm.Session 的子类
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.commit()

官方教程

https://github.com/fastapi/sqlmodel
https://sqlmodel.tiangolo.com/learn/
https://sqlmodel.fastapi.org.cn/learn/

简单使用样例

import datetime

from sqlmodel import Field, BigInteger, Integer, String, Boolean, DateTime, JSON, Session, insert, update, select, delete
from sqlmodel import SQLModel, create_engine


class BaseModel(SQLModel):
    id: int | None = Field(description="主键id", primary_key=True, sa_type=BigInteger,
                           sa_column_kwargs=dict(comment="主键id"))
    is_delete: bool = Field(default=False, description="是否逻辑删除", sa_type=Boolean,
                            sa_column_kwargs=dict(comment="是否逻辑删除"))
    updated_time: datetime.datetime | None = Field(description="更新时间", sa_type=DateTime,
                                                   sa_column_kwargs=dict(
                                                       comment="更新时间", onupdate=datetime.datetime.now,
                                                       server_onupdate=""))
    created_time: datetime.datetime | None = Field(default_factory=datetime.datetime.now, description="创建时间",
                                                   sa_type=DateTime,
                                                   sa_column_kwargs=dict(comment="创建时间"))


class User(BaseModel, table=True):  # 只有 table=True 的 model 会被创建为表
    __tablename__ = "tb_user"
    # 两个 32，一个用于 SQLAlchemy，一个用于 Pydantic
    name: str = Field(sa_type=String(32), max_length=32, sa_column_kwargs=dict(comment="用户名"))
    age: int = Field(sa_type=Integer, default=0, ge=0, le=200, sa_column_kwargs=dict(comment="年龄"))
    addresses: list[str] = Field(sa_type=JSON, default=[], sa_column_kwargs=dict(comment="地址列表"))
    other: dict = Field(sa_type=JSON, default={}, sa_column_kwargs=dict(comment="附加信息"))


engine = create_engine("mysql+pymysql://root:mYsql123456_@127.0.0.1:3306/dev")
SQLModel.metadata.create_all(engine)
session = Session(engine)

with session.begin():
    entity = User(name="张三", age=16, addresses=["beijing", "shanghai"])
    session.add(entity)

with session.begin():
    entity = User(name="张三", age=16, addresses=["beijing", "shanghai"])
    statement = insert(User).values(entity.model_dump(exclude_none=True))  # 使用pydantic转换成字典(排除None是避免id传入None报错)
    session.exec(statement)

with session.begin():
    entity = User(id=5, name="赵四", age=17, addresses=["beijing"])
    statement = update(User).where(User.id == 5).values(entity.model_dump())
    session.exec(statement)

with session.begin():
    statement = select(User).where(User.name == "赵四")
    result = session.exec(statement).all()
    print(result)

with session.begin():
    statement = delete(User).where(User.id == 6)
    session.exec(statement)

以使用者的角度看，SQLModel 整体上对 Pydantic 和 SQLAlchemy 有了一个大致封装，但是目前 v0.0.22 版本中有很多细节感觉做的不够细致导致有一种Pydantic 和 SQLAlchemy 简单缝合的感觉，比如一些类型提示不兼容导致静态检查报错，一些 SQLModel 中的 SQLALchemy 参数比较含糊，需要到 SQLAlachemy 中寻找具体参数进行补充。不过不影响具体执行。Python 生态中需要一个 view model 与 ORM model 统一的框架，静待这个库成熟。