对于一名新手程序员来说，学习数据库操作最快的方法是理解其三个核心层次：
- 数据库级操作：如何连接、创建或删除整个数据库。
- 表级操作：如何在数据库中创建或删除表的结构。
- 数据级操作：如何对表中的数据进行增、删、改、查 (CRUD)。
SQLModel 是一个现代化的 Python ORM 库，它能让你用 Python 代码优雅地完成上述大部分任务，尤其是表和数据级的操作。
第一部分：数据库级操作
ORM 的主要工作不是管理数据库服务器，而是与一个已经存在的数据库进行交互。因此，连接数据库是核心，而创建和删除通常是环境准备工作。
1. 连接数据库
在 SQLModel 中，“选择”一个数据库意味着创建一个指向它的“引擎”(Engine)。这是通过一个特殊的连接字符串 (URL) 完成的。
-
API:
create_engine() -
连接字符串格式:
数据库类型://用户名:密码@主机:端口/数据库名
from sqlmodel import create_engine
# 示例 1: 连接到一个本地的 PostgreSQL 数据库，名为 "mydb"
# 需要先安装 psycopg2: pip install psycopg2-binary
pg_engine = create_engine("postgresql://myuser:mypassword@localhost/mydb")
# 示例 2: 连接到一个 SQLite 数据库文件
# SQLite 是一个文件型数据库，如果文件不存在，会在首次操作时自动创建
sqlite_engine = create_engine("sqlite:///database.db")关键点
你的所有后续操作都将通过这个
engine对象在指定的数据库上执行。
2. 创建数据库
- 对于 SQLite: 如上所示，当你使用
create_engine指向一个不存在的文件时，该文件（即数据库）会被自动创建。 - 对于 PostgreSQL: 推荐在应用程序之外，使用数据库管理工具手动创建。也可以在终端运行
createdb my_new_database。
3. 删除数据库
- 对于 SQLite: 直接删除数据库文件即可。
import os if os.path.exists("database.db"): os.remove("database.db") - 对于 PostgreSQL: 同样，推荐使用管理工具。或在终端运行
dropdb my_database_to_delete
第二部分：表级操作
这是 SQLModel 强大之处。你通过定义 Python 类来定义数据表的结构。SQLModel 的设计强制要求使用类型提示语法来定义字段，但是对于初学者，完全可以在理解上暂时忽略 Pydantic 的数据校验功能和类型提示的深层含义。我们可以把冒号和后面的类型当作一个固定的写法，这个写法只有一个目的：告诉数据库，这一列应该是什么类型。
1. 定义表模型
每个继承了 SQLModel 的类，如果设置 table=True，就对应数据库中的一张表。
from typing import Optional
from sqlmodel import Field, SQLModel
class Hero(SQLModel, table=True):
# 类名 "Hero" 会被自动转换为表名 "hero"
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None)- 这里的
id、name、secret_name、age对应数据库中的列名。 :后面的内容指定列的类型。比如Optional[int]: 我们告诉数据库，这一列要存整数，并且可以为空。str: 我们告诉数据库，这一列要存文本。
Field()的作用：为列提供额外的元数据，这些元数据会被 SQLModel 用来精确地配置数据库表中对应列的 schema（结构、约束和行为）。
虽然 Field()有很多参数，但在日常开发中，只需要掌握几个最核心的就足够了。
| 参数 (Parameter) | 作用 (Function) | 使用场景示例 (Example Use Case) |
|---|---|---|
primary_key=True | 将列设为主键，作为每行数据的唯一标识。 | 用于每张表的唯一 ID |
default=... | 设置一个静态的默认值。 | 新用户默认为激活状态 |
unique=True | 保证该列中的所有值都是唯一的。 | 确保每个用户的邮箱都是唯一的 |
index=True | 为该列创建数据库索引以加快查询速度。 | 如果经常需要按名字搜索 |
foreign_key="..." | 将列设为外键，关联到另一张表的列。 | 在文章表中关联作者 |
max_length=... | 限制字符串类型的最大长度。 | 限制用户名长度 |
default_factory=... | 提供一个函数，用于动态生成默认值。 | 记录精确的创建时间 |
nullable=... | 显式声明该列是否可以为 NULL（空）。 | 通常由 Optional[...] 类型自动推断，很少需要手动设置 |
2. 创建表
SQLModel 会收集所有你定义的模型，然后你可以用一条命令在数据库中创建所有对应的表。
- API:
SQLModel.metadata.create_all(engine)
安全的命令
这个命令是安全的，可以重复执行。它会先检查表是否存在，只创建那些不存在的表。
# (接上文的 Hero 模型和 sqlite_engine)
def create_tables():
print("开始创建表...")
# 这条命令会连接到数据库，并创建 Hero 表
SQLModel.metadata.create_all(sqlite_engine)
print("表创建完成。")
# 在你的应用启动时调用一次即可
create_tables()3. 删除表
- API:
SQLModel.metadata.drop_all(engine)
危险操作
这是一个危险的操作，它会删除所有定义的表以及表中的全部数据。通常只在测试或开发初期重置数据库时使用。
def delete_tables():
print("警告：即将删除所有表...")
SQLModel.metadata.drop_all(sqlite_engine)
print("所有表已删除。")关于修改表结构 (数据库迁移)
如果你的 Hero 模型已经创建了对应的表，之后你又给模型增加了一个新字段（比如 hometown: str），再次运行 create_all 是不会更新表结构的。这时你需要专业的数据库迁移工具，例如 Alembic。对于初学者，可以先了解这个概念，在项目需要时再深入学习。
第三部分：数据级操作 (增删改查 - CRUD)
这是日常开发中最频繁的操作。所有这些操作都在一个“会话”(Session) 中进行。Session 对象是用来管理事务（Transaction）的。最常使用的·with Session(engine) as session: 代码块，其整体就代表了一个数据库事务。
准备工作
在我们可以对数据进行增删改查之前，必须先完成两项基础准备工作：
- 建立与数据库的连接：通过
create_engine()创建一个引擎 engine。 - 确保数据表已存在：通过
SQLModel.metadata.create_all(engine)确保代码中定义的模型（如 Hero）已经在数据库中创建了对应的表。
只有在这两项准备工作完成后，我们才能开始创建 Session 对象来执行具体的数据操作。
from sqlmodel import Session
# engine 来自第一部分
engine = create_engine("sqlite:///database.db")
# 创建 Hero 表
SQLModel.metadata.create_all(engine)1. 创建 (Create) - 增加数据
API: session.add(), session.commit(), session.refresh()
def create_heroes():
hero_1 = Hero(name="蜘蛛侠", secret_name="彼得·帕克", age=18)
hero_2 = Hero(name="钢铁侠", secret_name="托尼·斯塔克", age=45)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.commit() # commit() 才真正将数据写入数据库
# refresh() 可以用数据库中的新数据（如自动生成的id）更新Python对象
session.refresh(hero_1)
session.refresh(hero_2)
print("创建的英雄:", hero_1)
print("创建的英雄:", hero_2)
create_heroes()2.读取 (Read) - 查询数据
基本步骤
- 调用
select()以及它的链式方法得到一个Select对象（可以理解为SQL 语句的抽象表示，这个时候所有的操作都只是在 Python 内存中，没有与数据库进行任何通信） - 通过
session.exec()函数把这个Select对象交给session, session 会把这个 Python 对象翻译成真正的 SQL 语句，然后通过数据库连接发送出去并执行。最终得到一个Result对象。
from sqlmodel import select
def read_heroes():
with Session(engine) as session:
# 查询所有英雄
statement = select(Hero)
all_heroes = session.exec(statement).all()
print("所有英雄:", all_heroes)
# 按条件查询 (查询年龄大于20岁的英雄)
statement_filtered = select(Hero).where(Hero.age > 20)
older_heroes = session.exec(statement_filtered).all()
print("年龄大于20的英雄:", older_heroes)
# 查询单个英雄
statement_one = select(Hero).where(Hero.name == "蜘蛛侠")
spiderman = session.exec(statement_one).first()
if spiderman:
print("找到的蜘蛛侠:", spiderman)
read_heroes()select（）常用链式方法
| 方法 (Method) | 作用 (Function) | 使用场景示例 |
|---|---|---|
.where(...) | 过滤数据，只选择满足特定条件的行。这是最常用的方法。 | 选择所有年龄大于 20 岁的英雄，或组合多个条件进行筛选。 |
.order_by(...) | 排序结果，可以按升序（默认）或降序排列。 | 将查询结果按年龄从小到大或从大到小排序。 |
.limit(...) | 限制数量，只获取指定数量的结果。 | 只获取查询结果的前 5 名英雄。 |
.offset(...) | 跳过结果，从指定位置开始获取数据，常用于分页。 | 用于分页功能，例如跳过前 20 条记录，再获取之后的 10 条。 |
.join(...) | 连接表格，根据外键关系将两个或多个表的数据合并在一起查询。 | 在查询英雄的同时，获取其所在战队的名称等信息。 |
.group_by(...) | 分组数据，将具有相同值的行组合在一起，通常与聚合函数配合使用。 | 统计每个战队分别有多少名英雄。 |
.distinct() | 去重结果，确保返回的每一行都是唯一的。 | 获取所有英雄出现过的年龄列表，并去除重复的年龄值。 |
Result 对象常用方法
| 方法/操作 (Method/Operation) | 返回什么 (Returns What) | 使用场景示例 |
|---|---|---|
.all() | 一个列表，包含所有结果行。 | 获取所有结果，并且结果集不大时。 |
.first() | 单个结果对象，或者 None。 | 只需要第一条结果，或者确认结果最多只有一条时。 |
.one() | 单个结果对象。（若结果不唯一则报错） | 严格确认结果有且仅有一条时，用于数据完整性检查。 |
.one_or_none() | 单个结果对象，或者 None。（若结果多于一条则报错） | 确认结果最多只有一条，但允许为空。比 .one() 更安全。 |
for item in result: | 在循环的每一次迭代中返回单个结果对象。 | 逐一处理结果，尤其是在结果集很大时，可以节省内存。 |
.scalars() | 一个新的可迭代对象，它只会产生每行的第一列数据。 | 当你只查询一个字段（如所有英雄的名字）并想要一个干净的值列表时。 |
.mappings() | 一个新的可迭代对象，它会产生类似字典的行。 | 需要将结果转换为键值对格式，例如用于生成 JSON API 响应。 |
3. 更新 (Update) - 修改数据
流程：先查询，再修改对象的属性，最后提交。
- API: (查询) + 修改属性 +
session.add()+session.commit()
def update_hero():
with Session(engine) as session:
# 1. 找到要更新的英雄
statement = select(Hero).where(Hero.name == "蜘蛛侠")
hero_to_update = session.exec(statement).first()
if hero_to_update:
# 2. 修改其属性
hero_to_update.age = 19
# 3. 添加到会话并提交
session.add(hero_to_update)
session.commit()
session.refresh(hero_to_update)
print("更新后的英雄:", hero_to_update)
update_hero()4. 删除 (Delete) - 删除数据
流程：先查询，再从会话中删除，最后提交。
- API: (查询) +
session.delete()+session.commit()
def delete_hero():
with Session(engine) as session:
# 1. 找到要删除的英雄
statement = select(Hero).where(Hero.name == "钢铁侠")
hero_to_delete = session.exec(statement).first()
if hero_to_delete:
# 2. 从会话中删除
session.delete(hero_to_delete)
# 3. 提交
session.commit()
print("已删除英雄:", hero_to_delete.name)
delete_hero()