Python SQLModel 简单使用

简介

Documentation: https://sqlmodel.tiangolo.com

Source Code: https://github.com/fastapi/sqlmodel

SQLModel 是 Python 中用于 SQL 数据库的操作工具，旨在实现简洁性、兼容性和健壮性。

SQLModel 是基于 SQLAlchemy 和 Pydantic 构建的。它由 FastAPI 的同一作者制作，旨在完美匹配需要使用 SQL 数据库的 FastAPI 应用程序。

SQLModel 基于 Python 的类型注解，并由 Pydantic 和 SQLAlchemy 提供支持。

其主要特性包括：

• 易于编写：代码直观，编辑器支持良好，自动补全无处不在，减少调试时间。设计上易于使用和学习，节省阅读文档的时间。

• 使用简单：提供合理的默认值，在底层完成大量工作，从而简化你编写的代码。

• 兼容性强：专为与 FastAPI、Pydantic 和 SQLAlchemy 兼容而设计。

• 可扩展性强：底层拥有 SQLAlchemy 和 Pydantic 的全部功能和灵活性。

• 代码简洁：最大限度减少代码重复。一条类型注解即可完成多项任务，无需在 SQLAlchemy 和 Pydantic 中重复定义模型。

安装 SQLModel

由于 SQLModel 基于 Pydantic 和 SQLAlchemy 构建，因此它依赖这两个库。当你安装 SQLModel 时，它们将会被自动安装。

pip install sqlmodel -i https://mirrors.aliyun.com/pypi/simple/pip install pymysql -i https://mirrors.aliyun.com/pypi/simple/

创建数据模型

我们要做的第一件事是创建一个类来表示数据表中的数据。这种代表某些数据的类通常被称为模型（model）。

from typing import Optionalfrom sqlmodel import SQLModel, Fieldclass Users(SQLModel, table=True):__tablename__ = 'users'id: str = Field(primary_key=True, description="主键ID", max_length=64)name: str = Field(description="用户名", max_length=50)age: Optional[int] = Field(description="年龄", default=None)

table=True 会告诉 SQLModel 这是一个表模型，它应该表示 SQL 数据库中的一个表，而不仅仅是一个数据模型（就像其他常规的 Pydantic 类一样）。

创建引擎（Engine）

SQLModel 的引擎 engine（实际上它是一个 SQLAlchemy engine ）是用来与数据库保持连接的。

创建引擎非常简单，只需使用一个数据库连接 URL 调用 create_engine() 即可：

from sqlmodel import create_engine# 连接到 MySQL 数据库
username = 'root'
password = '666666'
host = '127.0.0.1'  # 例如：'localhost' 或 '127.0.0.1'
port = '3306'  # 通常是 3306
database = 'my_sqlalchemy'
# 创建连接引擎
dbHost = f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'engine = create_engine(dbHost,echo=True,  # 是否打印SQLpool_size=10,  # 连接池的大小，指定同时在连接池中保持的数据库连接数，默认:5max_overflow=20,  # 超出连接池大小的连接数，超过这个数量的连接将被丢弃,默认: 5
)

通常情况下，你的整个应用程序中应该只创建一个引擎对象，并在各处重复使用它。

你可以在 SQLAlchemy 官方文档 中了解更多关于它所支持的各种数据库（同时也是 SQLModel 所支持的数据库）的详细信息。

SQLModel 为其 create_engine() 函数提供了自己的定义。它与 SQLAlchemy 的 create_engine() 是相同的，只是默认设置了 future=True 参数（这意味着它使用了 SQLAlchemy 最新版本，即 1.4 和未来的 2.0 所采用的风格）。

创建会话 Session

from sqlmodel import Session# 创建Session会话
session = Session(engine)

Session 以一个引擎（engine）作为参数，并在其底层使用该引擎。

添加数据

from data_model.users import Users
from mysql_client import sessionuser1 = Users(id="1", name="tom")
user2 = Users(id="2", name="jetty", age=18)
user3 = Users(id="3", name="tonny", age=22)session.add(user1)
session.add(user2)
session.add(user3)session.commit()

使用 with 语句块创建Session

from sqlmodel import Sessionfrom data_model.users import Users
from mysql_client import engineuser1 = Users(id="1", name="tom")
user2 = Users(id="2", name="jetty", age=18)
user3 = Users(id="3", name="tonny", age=22)with Session(engine) as session:session.add(user1)session.add(user2)session.add(user3)session.commit()

使用 with 语句块后，会在进入 with 块时自动创建会话，并将其赋值给变量 session，而在 with 块执行完毕后，会话会自动关闭。

查询数据

select语句

def select_all():with Session(engine) as session:statement = select(Users) # SQLModel的select语句

statement = select(Users) 相当于SQL语句：select id, name, age from users

执行该语句

现在我们已经有了 select 语句，可以使用会话（session）来执行它：

def select_all():with Session(engine) as session:statement = select(Users) # SQLModel的select语句result = session.exec(statement) # 执行

遍历查询结果

results 对象是一个可迭代对象，可以用来逐行遍历查询结果中的每一条记录。

现在我们可以将它放入一个 for 循环中，逐个打印出信息：

def select_all():with Session(engine) as session:statement = select(Users) # SQLModel的select语句result = session.exec(statement) # 执行for user in result:print(user)

完整代码：

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginedef select_all():with Session(engine) as session:statement = select(Users) # SQLModel的select语句result = session.exec(statement) # 执行for user in result:print(user)if __name__ == '__main__':select_all()

运行结果：

2025-06-21 22:50:10,074 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2025-06-21 22:50:10,074 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:50:10,075 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2025-06-21 22:50:10,075 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:50:10,076 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2025-06-21 22:50:10,076 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:50:10,077 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-06-21 22:50:10,082 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.age 
FROM users
2025-06-21 22:50:10,082 INFO sqlalchemy.engine.Engine [generated in 0.00028s] {}
name='tom' id='1' age=None
name='jetty' id='2' age=18
name='tonny' id='3' age=22
2025-06-21 22:50:10,084 INFO sqlalchemy.engine.Engine ROLLBACK

获取对象列表

使用 result.all()，它可以返回包含所有对象的列表：

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginedef select_all():with Session(engine) as session:statement = select(Users) # SQLModel的select语句result = session.exec(statement) # 执行users = result.all() # 获取list对象print(users)if __name__ == '__main__':select_all()

运行结果：

2025-06-21 22:54:35,797 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2025-06-21 22:54:35,797 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:54:35,798 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2025-06-21 22:54:35,798 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:54:35,799 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2025-06-21 22:54:35,799 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:54:35,799 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-06-21 22:54:35,801 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.age 
FROM users
2025-06-21 22:54:35,802 INFO sqlalchemy.engine.Engine [generated in 0.00014s] {}
[Users(name='tom', id='1', age=None), Users(name='jetty', id='2', age=18), Users(name='tonny', id='3', age=22)]
2025-06-21 22:54:35,803 INFO sqlalchemy.engine.Engine ROLLBACK

简洁版本

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginedef select_all():with Session(engine) as session:users = session.exec(select(Users)).all()print(users)if __name__ == '__main__':select_all()

运行结果：

2025-06-21 22:57:54,214 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2025-06-21 22:57:54,215 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:57:54,216 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2025-06-21 22:57:54,216 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:57:54,216 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2025-06-21 22:57:54,216 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-06-21 22:57:54,217 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-06-21 22:57:54,220 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.age 
FROM users
2025-06-21 22:57:54,220 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {}
[Users(name='tom', id='1', age=None), Users(name='jetty', id='2', age=18), Users(name='tonny', id='3', age=22)]
2025-06-21 22:57:54,222 INFO sqlalchemy.engine.Engine ROLLBACK

where 条件查询

使用 SQLModel 通过 WHERE 子句过滤行

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginedef select_user():with Session(engine) as session:stmt = select(Users).where(Users.name == 'tom')results = session.exec(stmt)for user in results:print(user)if __name__ == '__main__':select_user()

可以连续调用where子句。

stmt = select(Users).where(Users.name == 'tom').where(Users.age > 18)

where 条件操作符：<、>、==、>=、<= 和 != 都是用于比较的运算符。

使用 or_

使用 or_ 在 .where() 中组合多个表达式，要做到这一点，你可以导入 or_：

from sqlmodel import Field, Session, SQLModel, create_engine, or_, selectfrom sqlmodel import Session, select, or_from data_model.users import Users
from mysql_client import enginedef select_user():with Session(engine) as session:stmt = select(Users).where(or_(Users.age > 18, Users.age < 60))results = session.exec(stmt)for user in results:print(user)if __name__ == '__main__':select_user()

读取单行数据

.first() 方法

读取第一行数据

但假设我们并不关心所有的行，而只想要第一行数据。我们可以调用 .first() 方法来获取第一行数据

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import engineif __name__ == '__main__':with Session(engine) as session:stmt = select(Users).where(Users.age > 18)result = session.exec(stmt)user = result.first() # 读取第一行数据print(user)

查询为空的情况下返回：None

.one() 方法

精确匹配一行（Exactly One）

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import engineif __name__ == '__main__':with Session(engine) as session:stmt = select(Users).where(Users.name == 'tom')result = session.exec(stmt)user = result.one()print(user)

结果返回了多行数据，但又期望只匹配一行时使用 .one()，会引发错误：

Traceback (most recent call last):File "D:\17-SQLModel\project\SQLModel\05-读取单行数据-one方法.py", line 16, in <module>user = result.one()^^^^^^^^^^^^File "D:\Miniconda\miniconda3\envs\SQLModel\Lib\site-packages\sqlalchemy\engine\result.py", line 1815, in onereturn self._only_one_row(^^^^^^^^^^^^^^^^^^^File "D:\Miniconda\miniconda3\envs\SQLModel\Lib\site-packages\sqlalchemy\engine\result.py", line 813, in _only_one_rowraise exc.MultipleResultsFound(
sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

同样地，如果使用 .one() 却完全没有查询到任何行，它也会抛出一个错误：

Traceback (most recent call last):File "D:\17-SQLModel\project\SQLModel\05-读取单行数据-one方法.py", line 16, in <module>user = result.one()^^^^^^^^^^^^File "D:\Miniconda\miniconda3\envs\SQLModel\Lib\site-packages\sqlalchemy\engine\result.py", line 1815, in onereturn self._only_one_row(^^^^^^^^^^^^^^^^^^^File "D:\Miniconda\miniconda3\envs\SQLModel\Lib\site-packages\sqlalchemy\engine\result.py", line 760, in _only_one_rowraise exc.NoResultFound(
sqlalchemy.exc.NoResultFound: No row was found when one was required

简洁写法

当然，在大多数情况下，使用 .first() 和 .one() 时你可能也会倾向于用更简洁的形式来书写，通常是一行代码（或至少是一个完整的 Python 语句）：

user = session.exec(select(Users).where(Users.name == "Deadpond")).one()

Select by Id with .where()

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import engineif __name__ == '__main__':with Session(engine) as session:user = session.exec(select(Users).where(Users.id == 1)).first()print(user)

Select by Id with .get()

由于通过主键的 Id 列选择单个行是一种常见操作，因此这里有一个快捷方式：

from sqlmodel import Sessionfrom data_model.users import Users
from mysql_client import enginewith Session(engine) as session:user = session.get(Users, 11)print(user)

数据查询为空的情况下返回: None

limit 和 offset

limit查询

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginewith Session(engine) as session:stmt = select(Users).limit(3) # 从结果集中获取前3条数据result = session.exec(stmt)users = result.all()print(users)

简洁写法：

users = session.exec(select(Users).limit(3)).all()

Offset and Limit

from sqlmodel import Session, selectfrom data_model.users import Users
from mysql_client import enginewith Session(engine) as session:stmt = select(Users).where(Users.age < 100).offset(0).limit(3).order_by(Users.id)users = session.exec(stmt).all()print(users)

修改/更新数据

from sqlmodel import Sessionfrom data_model.users import Users
from mysql_client import enginewith Session(engine) as session:user = session.get(Users, 1) # 根据ID查询user.age = 100 # 设置字段值session.add(user) # 添加session.commit() # 提交session.refresh(user) # 刷新数据print(user)  # 打印

运行结果：

2025-06-22 16:20:22,916 INFO sqlalchemy.engine.Engine BEGIN (implicit)
# 查询
2025-06-22 16:20:22,919 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.age AS users_age 
FROM users 
WHERE users.id = %(pk_1)s
2025-06-22 16:20:22,920 INFO sqlalchemy.engine.Engine [generated in 0.00021s] {'pk_1': 1}
# 更新
2025-06-22 16:20:22,923 INFO sqlalchemy.engine.Engine UPDATE users SET age=%(age)s WHERE users.id = %(users_id)s
2025-06-22 16:20:22,923 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {'age': 100, 'users_id': '1'}
2025-06-22 16:20:22,924 INFO sqlalchemy.engine.Engine COMMIT
2025-06-22 16:20:22,927 INFO sqlalchemy.engine.Engine BEGIN (implicit)
# 刷新
2025-06-22 16:20:22,928 INFO sqlalchemy.engine.Engine SELECT users.id, users.name, users.age 
FROM users 
WHERE users.id = %(pk_1)s
2025-06-22 16:20:22,929 INFO sqlalchemy.engine.Engine [generated in 0.00015s] {'pk_1': '1'}
name='tom' id='1' age=100
2025-06-22 16:20:22,929 INFO sqlalchemy.engine.Engine ROLLBACK

删除数据

from sqlmodel import Sessionfrom data_model.users import Users
from mysql_client import enginewith Session(engine) as session:user = session.get(Users, 1)if user:session.delete(user) # 删除session.commit()

join 表连接

看官方文档吧！

FastAPI整合SQLModel

SQLModel 最为出色的一个使用场景，也是它被创建的主要原因之一，就是与 FastAPI 结合使用。

FastAPI 是一个由 SQLModel 的作者所创建的 Python Web 框架，专门用于构建 Web API。FastAPI 同样是基于 Pydantic 构建的。

安装SQLModel

pip install sqlmodel -i https://mirrors.aliyun.com/pypi/simple/pip install pymysql -i https://mirrors.aliyun.com/pypi/simple/

创建数据模型

base_sql_model.py 数据模型基类

from datetime import datetimefrom sqlmodel import SQLModel, Fieldclass BaseSQLModel(SQLModel):id: str = Field(primary_key=True, description="主键ID", max_length=64)create_time: datetime = Field(description="创建时间", default_factory=datetime.now)update_time: datetime = Field(description="修改时间", default_factory=datetime.now)is_delete: int = Field(description="是否删除:0-未删除，1-已删除", max_length=1, default=0)

user_model.py 数据模型类

from typing import Optionalfrom sqlmodel import Fieldfrom apps.user.data_model.base_sql_model import BaseSQLModelclass User(BaseSQLModel, table=True):__tablename__ = 'user'username: str = Field(description="用户名", max_length=50)password: str = Field(description="密码", max_length=100)phone: Optional[int] = Field(description="手机号", default=0)status: int = Field(description="用户状态:0-可用，1-不可用", max_length=1, default=0)

创建引擎以及Session会话

mysql_clinet.py

from sqlmodel import create_engine, Sessionfrom config import app_settingsclass MySQLClient:username = app_settings.mysql.usernamepassword = app_settings.mysql.passwordhost = app_settings.mysql.host  # 例如：'localhost' 或 '127.0.0.1'port = app_settings.mysql.port  # 通常是 3306database = app_settings.mysql.database_namedbHost = f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'engine = create_engine(dbHost,echo=app_settings.mysql.echo_sql,  # 是否打印SQLpool_size=app_settings.mysql.pool_size,  # 连接池的大小，指定同时在连接池中保持的数据库连接数，默认:5max_overflow=app_settings.mysql.max_overflow,  # 超出连接池大小的连接数，超过这个数量的连接将被丢弃,默认: 5pool_timeout=app_settings.mysql.pool_timeout,pool_recycle=app_settings.mysql.pool_recycle,pool_pre_ping=app_settings.mysql.pool_pre_ping,)@classmethoddef get_db_session(cls,)-> Session:with Session(cls.engine) as session:yield session

FastAPI中使用SQLModel

user_router.py

users = APIRouter()@users.post("/register", response_model=Result, summary="用户注册")
async def register(user_register: UserRegister, user_service: UserService = Depends()):user_service.register(user_register)return Result(message="注册成功,请跳转到登录")

user_service.py

class UserService:def __init__(self,user_dao: UserDAO = Depends(),user_ext_info_dao: UserExtInfoDAO = Depends(),redis_client: RedisClient = Depends(get_redis_client),db_session: Session = Depends(MySQLClient.get_db_session),):self.user_dao = user_daoself.user_ext_info_dao = user_ext_info_daoself.redis_client = redis_clientself.db_session = db_sessiondef find_by_user_id(self, user_id)-> UserVo:"""根据用户ID查询"""user = self.user_dao.find_by_id(user_id)# 用户不存在if user is None:raise UserNotFoundException# 字典解包user_vo = UserVo(**user.__dict__)return user_vo

user_dao.py

class UserDAO:def __init__(self, db_session: Session = Depends(MySQLClient.get_db_session),):self.db_session = db_sessiondef find_by_id(self, user_id):return self.db_session.get(User, user_id)def find_user_by_username(self, username: str):