3.1 增：插入数据
CRUD 的第一课。本节学会：单条插入、批量插入、拿回自增主键、处理唯一约束冲突。
本章公共代码
第 3 章所有示例基于同一个模型，先建好 models.py：
# models.py
from datetime import datetime
from typing import Optional
from sqlalchemy import String, create_engine, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[str] = mapped_column(String(120), unique=True)
age: Mapped[Optional[int]]
city: Mapped[Optional[str]] = mapped_column(String(50))
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
def __repr__(self) -> str:
return f"User(id={self.id}, name={self.name!r}, age={self.age}, city={self.city!r})"
engine = create_engine("sqlite:///crud.db", echo=True)
SessionLocal = sessionmaker(bind=engine)
Base.metadata.create_all(engine)
之后的示例文件统一这样开头：
from models import User, SessionLocal
一、插入一条数据：三步走
from models import User, SessionLocal
with SessionLocal() as session:
# ① 创建对象
user = User(name="张三", email="zhangsan@example.com", age=25, city="北京")
# ② 加入会话
session.add(user)
# ③ 提交
session.commit()
print(f"插入成功，分配到的 id = {user.id}")
对应生成的 SQL（echo 日志里能看到）：
INSERT INTO users (name, email, age, city) VALUES (?, ?, ?, ?)
commit 之后发生了什么？
- 数据真正写入数据库
user.id自动被填上数据库分配的自增主键——不需要你做任何事created_at这类server_default字段也由数据库填好了
💡 如果 commit 后访问
user.created_at，SQLAlchemy 会自动发一条 SELECT 把数据库生成的值取回来（因为 commit 后对象会"过期"，首次访问属性时自动刷新）。
二、批量插入：add_all
with SessionLocal() as session:
users = [
User(name="李四", email="lisi@example.com", age=30, city="上海"),
User(name="王五", email="wangwu@example.com", age=17, city="北京"),
User(name="赵六", email="zhaoliu@example.com", age=42, city="深圳"),
User(name="孙七", email="sunqi@example.com", age=28, city="上海"),
]
session.add_all(users)
session.commit()
一次 commit 提交多条是常规做法：所有 INSERT 在同一个事务里，要么全部成功、要么全部失败，而且比逐条 commit 快得多。
# ❌ 慢且不原子：每条一个事务
for u in users:
session.add(u)
session.commit()
# ✅ 快且原子：一个事务搞定
session.add_all(users)
session.commit()
三、超大批量：bulk insert
要灌入几万条数据（导数据、造测试数据）时，走 ORM 对象会比较慢。用 Core 风格的 insert() 直接批量执行：
from sqlalchemy import insert
from models import User, SessionLocal
data = [
{"name": f"用户{i}", "email": f"user{i}@example.com", "age": 20 + i % 30}
for i in range(10000)
]
with SessionLocal() as session:
session.execute(insert(User), data) # 一次性批量 INSERT
session.commit()
| 方式 | 特点 | 适用 |
|---|---|---|
add_all + ORM 对象 | 走完整 ORM 流程（默认值、事件、拿回主键） | 日常业务，几十几百条 |
session.execute(insert(User), 字典列表) | 跳过对象管理，速度快数倍 | 数据导入，上万条 |
四、处理唯一约束冲突
email 列有 unique=True，插入重复邮箱会抛异常：
from sqlalchemy.exc import IntegrityError
from models import User, SessionLocal
with SessionLocal() as session:
user = User(name="张三二号", email="zhangsan@example.com") # 邮箱已存在！
session.add(user)
try:
session.commit()
except IntegrityError:
session.rollback() # 必须回滚，否则这个 session 无法继续使用
print("邮箱已被注册！")
两个要点：
- 捕获的是
IntegrityError（完整性错误：唯一约束、非空约束、外键约束违反都抛它） - except 里必须
rollback()——commit 失败后 Session 处于"故障"状态，不回滚的话后续任何操作都会报PendingRollbackError
更友好的做法：先查后插
from sqlalchemy import select
with SessionLocal() as session:
email = "zhangsan@example.com"
exists = session.scalar(select(User).where(User.email == email))
if exists:
print("邮箱已被注册")
else:
session.add(User(name="张三", email=email))
session.commit()
📌 注意"先查后插"在高并发下仍有极小概率撞车（两个请求同时通过了检查），所以唯一约束 + 捕获 IntegrityError 是最后防线，两者都要有。
五、插入时的默认值回顾
user = User(name="张三", email="zs@example.com")
# age、city 没传 → NULL（因为是 Optional）
# created_at 没传 → 数据库自动填当前时间（server_default）
# id 没传 → 数据库自增分配
不要手动给 id 赋值，交给数据库自增管理。
六、常见疑问
Q：add 之后、commit 之前，能查到这条数据吗？ 在同一个 session 里可以（SQLAlchemy 查询前会自动 flush 待办的 INSERT）；其他 session / 其他程序看不到，因为事务还没提交。
Q：怎么在 commit 前就拿到自增 id？
session.flush()——SQL 发出去了（id 已分配），但事务未提交，还能反悔。
Q：忘了 commit 会怎样？
with 块结束时 session 关闭，未提交的改动全部丢弃。新手"为什么数据没存进去"的第一大原因就是忘了 commit。
📝 本节小结
- 插入三步：
User(...)→session.add()→session.commit() - commit 后自增
id自动回填到对象上 - 多条数据放同一个事务里一次 commit：更快、更安全
- 超大批量用
session.execute(insert(User), 字典列表) - 唯一约束冲突抛
IntegrityError，捕获后必须 rollback
下一节，学 CRUD 里最丰富的"查" → 3.2 查询数据