SQLModel：Python 中高效的数据库交互工具

作者：

在

Python 凭借其简洁的语法、丰富的生态以及强大的扩展性，在 Web 开发、数据分析、机器学习、自动化脚本等众多领域占据了重要地位。从金融领域的量化交易到科研机构的数据分析，从企业级 Web 应用到桌面自动化任务，Python 的身影无处不在。而在数据处理与存储的核心场景中，数据库交互是绕不开的关键环节。本文将聚焦于一款专为 Python 打造的高效数据库工具——SQLModel，深入解析其功能特性、使用方式及实际应用场景，帮助开发者轻松驾驭数据库操作。

一、SQLModel 概述：用途、原理与特性

1. 用途与定位

SQLModel 是一款基于 Python 的新型数据库 ORM（对象关系映射）工具，旨在简化数据库模型定义、查询构建及事务管理流程。它融合了 SQLAlchemy 的强大功能与 Pydantic 的数据验证特性，特别适合快速开发 API 服务、后端应用及需要复杂数据库交互的项目。无论是创建新的数据库表结构，还是执行复杂的 SQL 查询，SQLModel 都能通过 Python 代码实现无缝操作，极大降低了开发者与数据库打交道的门槛。

2. 工作原理

SQLModel 基于 SQLAlchemy 的核心引擎构建，底层依赖 SQLAlchemy 的 SQL 表达式生成器与数据库连接池。其核心逻辑在于通过 Python 类定义数据库模型（Model），这些类同时继承自 SQLModel 和 Pydantic.BaseModel，因此兼具 ORM 映射与数据验证功能。当定义模型类时，通过字段类型（如 Integer、String）与约束条件（如 primary_key=True、index=True）自动生成对应的数据库表结构；在执行查询时，SQLModel 将 Python 方法转换为 SQL 语句，并通过会话（Session）管理数据库连接与事务。

3. 核心优缺点

优点：

语法简洁：结合 Pydantic 的数据模型定义方式，代码可读性极高，减少样板代码。
类型安全：基于 Pydantic 的类型验证，确保数据完整性，提前捕获类型错误。
兼容性强：支持 SQLite、PostgreSQL、MySQL 等主流关系型数据库，切换数据库时只需修改连接字符串。
开发高效：内置自动生成 CRUD（增删改查）方法，支持异步操作（通过 AsyncSQLModel），适合 FastAPI 等异步框架。

缺点：

学习曲线：对于完全没有 SQLAlchemy 基础的开发者，需理解 ORM 概念及底层原理。
复杂查询限制：对于极复杂的原生 SQL 查询，可能需要结合 SQLAlchemy 的原生表达式或直接编写 SQL 语句。

4. License 类型

二、SQLModel 安装与基础使用

1. 环境准备与安装

依赖要求

Python 3.7+
目标数据库驱动（如 pymysql 用于 MySQL，psycopg2-binary 用于 PostgreSQL）

安装命令

# 安装 SQLModel（含 SQLite 驱动）
pip install sqlmodel

# 可选：安装其他数据库驱动
# MySQL: pip install pymysql
# PostgreSQL: pip install psycopg2-binary

2. 基础使用流程：定义模型与操作数据库

（1）定义数据库模型

from sqlmodel import SQLModel, Field, create_engine
from typing import Optional, List

# 定义用户模型
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)  # 主键，自动生成
    name: str = Field(index=True)  # 带索引的字符串字段
    email: str = Field(unique=True, index=True)  # 唯一且带索引
    age: Optional[int] = None  # 可选整数字段
    hobbies: Optional[List[str]] = None  # 存储列表（需数据库支持 JSON 类型）

关键点说明：

table=True：标识该类为数据库表模型，否则仅作为 Pydantic 数据模型使用。
Field 参数：设置字段约束，如 primary_key（主键）、index（索引）、unique（唯一）、default（默认值）等。
类型注解：直接使用 Python 原生类型（如 str、int）或 Pydantic 类型（如 EmailStr），自动映射数据库类型。

（2）创建数据库连接与表结构

# 创建 SQLite 数据库引擎（文件存储于当前目录）
engine = create_engine("sqlite:///test.db", echo=True)  # echo=True 打印 SQL 语句

# 创建所有表结构（基于模型定义）
SQLModel.metadata.create_all(engine)

说明：

create_engine：根据连接字符串创建数据库引擎，支持 SQLite、PostgreSQL、MySQL 等格式。
SQLModel.metadata.create_all(engine)：根据所有继承自 SQLModel 且 table=True 的模型类创建表。

（3）基本 CRUD 操作：使用会话（Session）

from sqlmodel import Session, select

# 创建会话（管理数据库连接与事务）
with Session(engine) as session:
    # 1. 创建数据（新增）
    user1 = User(name="Alice", email="[email protected]", age=28)
    session.add(user1)  # 添加到会话
    session.commit()  # 提交事务
    session.refresh(user1)  # 刷新对象，获取数据库生成的 ID
    print(f"Created user: {user1.id}, {user1.name}")

    # 2. 查询数据（单条与多条）
    # 查询单条（通过 ID）
    db_user = session.get(User, user1.id)
    print(f"Retrieved user: {db_user.name}")

    # 查询所有用户
    users = session.exec(select(User)).all()
    print(f"Total users: {len(users)}")

    # 3. 更新数据
    db_user.age = 30
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    print(f"Updated age: {db_user.age}")

    # 4. 删除数据
    session.delete(db_user)
    session.commit()
    print("User deleted successfully")

核心概念解析：

会话（Session）：SQLModel 通过会话管理数据库操作，所有增删改查需在会话中执行。
select 语句：使用 SQLModel 的 select 函数构建查询条件，避免拼接 SQL 字符串的安全隐患。
事务管理：commit() 提交事务，rollback() 回滚（未展示），确保数据一致性。

三、进阶功能与实战场景

1. 关系模型：一对一与一对多关联

（1）定义关联模型（以用户-地址为例）

# 定义地址模型（与用户一对一关联）
class Address(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    street: str
    city: str
    user_id: Optional[int] = Field(default=None, foreign_key="user.id")  # 外键关联用户表

    # 定义关联关系（可选，用于反向查询）
    user: Optional[User] = Relationship(back_populates="address")

# 更新用户模型，添加关联字段
class User(SQLModel, table=True):
    # ... 原有字段 ...
    address: Optional[Address] = Relationship(back_populates="user")  # 一对一关联

（2）创建关联数据

with Session(engine) as session:
    # 创建用户与地址
    user = User(name="Bob", email="[email protected]")
    address = Address(street="123 Main St", city="New York", user=user)

    session.add(address)  # 添加关联对象时，会自动处理用户的添加
    session.commit()
    session.refresh(user)
    print(f"User address: {user.address.city}")

（3）一对多关联（以用户-订单为例）

# 定义订单模型
class Order(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    amount: float
    user_id: int = Field(foreign_key="user.id")

    user: User = Relationship(back_populates="orders")  # 反向关联用户

# 更新用户模型，添加订单列表
class User(SQLModel, table=True):
    # ... 原有字段 ...
    orders: List[Order] = Relationship(back_populates="user")  # 一对多关联

关联查询示例：

# 查询用户及其所有订单
user = session.get(User, 1)
for order in user.orders:
    print(f"Order {order.id}: ${order.amount}")

2. 异步操作：支持 FastAPI 等异步框架

（1）定义异步模型

from sqlmodel import AsyncSQLModel, create_async_engine

class AsyncUser(AsyncSQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

# 创建异步引擎（以 PostgreSQL 为例）
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@host:port/db",
    echo=True
)

（2）异步 CRUD 操作

from sqlmodel import AsyncSession

async def create_user_async():
    async with AsyncSession(async_engine) as session:
        user = AsyncUser(name="Charlie")
        session.add(user)
        await session.commit()
        await session.refresh(user)
        print(f"Created async user: {user.id}")

# 运行异步函数
import asyncio
asyncio.run(create_user_async())

适用场景：

FastAPI 应用中使用 async def 定义路由，配合 SQLModel 异步会话实现非阻塞数据库操作。

3. 复杂查询：组合条件与原生 SQL

（1）条件查询（`where` 子句）

from sqlalchemy import and_, or_

# 查询年龄大于 25 且邮箱包含 "example" 的用户
statement = select(User).where(
    and_(User.age > 25, User.email.contains("example"))
)
users = session.exec(statement).all()

（2）原生 SQL 查询

# 执行原生 SQL（需注意防注入）
results = session.execute("SELECT * FROM user WHERE age > :age", {"age": 30})
for row in results:
    print(row.name)

注意事项：

原生 SQL 需通过 session.execute() 执行，返回结果为 Result 对象，可通过 .all() 或迭代获取数据。
避免直接拼接用户输入到 SQL 字符串中，始终使用参数化查询（如 :age 占位符）。

四、实际案例：构建用户管理 API（结合 FastAPI）

1. 项目结构

project/
├── main.py         # FastAPI 入口文件
├── models.py       # SQLModel 模型定义
└── database.py     # 数据库连接配置

2. 数据库配置（`database.py`）

from sqlmodel import create_engine, Session

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, echo=True)

def get_session():
    with Session(engine) as session:
        yield session

3. 模型定义（`models.py`）

from sqlmodel import SQLModel, Field
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    age: Optional[int] = None

4. FastAPI 路由（`main.py`）

from fastapi import FastAPI, Depends
from sqlmodel import Session, select
from models import User
from database import get_session, engine

# 创建表结构（启动时执行）
SQLModel.metadata.create_all(engine)

app = FastAPI()

# 新增用户
@app.post("/users/")
def create_user(user: User, session: Session = Depends(get_session)):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

# 查询所有用户
@app.get("/users/", response_model=list[User])
def read_users(session: Session = Depends(get_session)):
    users = session.exec(select(User)).all()
    return users

# 查询单个用户
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# 更新用户
@app.patch("/users/{user_id}")
def update_user(user_id: int, user_data: User, session: Session = Depends(get_session)):
    db_user = session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")

    # 更新字段（仅更新存在的参数）
    if user_data.name:
        db_user.name = user_data.name
    if user_data.email:
        db_user.email = user_data.email
    if user_data.age is not None:
        db_user.age = user_data.age

    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

# 删除用户
@app.delete("/users/{user_id}")
def delete_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    session.delete(user)
    session.commit()
    return {"message": "User deleted successfully"}

5. 启动与测试

（1）安装依赖

pip install fastapi uvicorn sqlmodel

（2）启动服务

uvicorn main:app --reload

（3）测试接口

通过 Swagger UI 访问：http://127.0.0.1:8000/docs
使用 curl 测试新增用户：

  curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"name":"David", "email":"[email protected]", "age":35}'

五、资源链接

1. PyPI 地址

https://pypi.org/project/sqlmodel

2. GitHub 地址

https://github.com/tiangolo/sqlmodel

3. 官方文档地址

https://sqlmodel.tiangolo.com

总结：SQLModel 为何值得选择？

SQLModel 通过融合 SQLAlchemy 的强大功能与 Pydantic 的开发体验，为 Python 开发者提供了一套简洁、高效且类型安全的数据库解决方案。无论是快速搭建 API 服务的原型，还是开发复杂的企业级应用，其自动生成 CRUD、无缝支持异步操作、灵活处理关联关系等特性都能显著提升开发效率。通过本文的实例演示，我们可以看到，从基础的单表操作到复杂的业务逻辑，SQLModel 都能以清晰的代码结构实现功能。对于正在寻找 ORM 工具的开发者，尤其是 FastAPI 用户，SQLModel 是值得优先考虑的选择。通过实践不同场景的代码示例，逐步掌握其核心逻辑，即可在数据库交互场景中发挥 Python 的最大效能。

关注我，每天分享一个实用的Python自动化工具。

实用工具