DevNull 知识库

Appearance

35｜SQLAlchemy 模型定义与 CRUD ​

直接写 SQL 能完成所有数据库操作，但代码中到处散落着 SQL 字符串，维护困难、容易写错、数据库迁移也麻烦。ORM（对象关系映射）把数据库表映射为 Python 类，把行映射为对象，用 Python 代码代替手写 SQL。SQLAlchemy 是 Python 最主流的 ORM，这篇从模型定义开始，把增删改查用 ORM 方式重写一遍。

一、安装与连接 ​

bash
uv add sqlalchemy
python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 创建引擎（SQLite 示例）
engine = create_engine("sqlite:///ops.db", echo=True)

# 基类，所有模型都继承它
Base = declarative_base()

# Session 工厂
SessionLocal = sessionmaker(bind=engine)
对象作用
Engine数据库连接池和方言管理
Base声明式基类，收集所有模型定义
Session数据库会话，封装事务边界

echo=True 会把所有执行的 SQL 打印到控制台，开发调试时有用，生产环境应关闭。

二、定义模型 ​

python
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

class Server(Base):
    __tablename__ = "servers"

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), nullable=False, unique=True)
    ip = Column(String(15), nullable=False)
    status = Column(String(20), default="running")
    created_at = Column(DateTime, default=datetime.utcnow)
列定义作用
Column(Integer, primary_key=True)整型主键
autoincrement=True自增
String(64)最大长度 64 的字符串
nullable=False不允许 NULL
unique=True唯一约束
default="running"默认值

模型定义完成后，创建表：

python
Base.metadata.create_all(bind=engine)

这会生成 CREATE TABLE SQL 并执行。表已存在时不会报错（类似 IF NOT EXISTS）。

三、CRUD 操作 ​

创建（Create） ​

python
session = SessionLocal()

new_server = Server(hostname="web-01", ip="192.168.1.10")
session.add(new_server)
session.commit()
session.refresh(new_server)   # 获取数据库生成的默认值（如 id、created_at）

print(new_server.id)   # 1

add() 把对象加入会话，commit() 提交事务，refresh() 从数据库重新加载对象（获取自增 ID 等）。

查询（Read） ​

python
from sqlalchemy import select

# 查询全部
stmt = select(Server)
results = session.execute(stmt).scalars().all()

# 按条件查询
stmt = select(Server).where(Server.status == "running")
results = session.execute(stmt).scalars().all()

# 查询单条
stmt = select(Server).where(Server.id == 1)
server = session.execute(stmt).scalar_one_or_none()

SQLAlchemy 2.0 推荐使用 select() 语法（类似 SQL 的 SELECT 语句），而不是旧的 query() API。

更新（Update） ​

python
server = session.get(Server, 1)   # 通过主键获取
server.status = "stopped"
session.commit()

直接修改对象属性，然后 commit()。SQLAlchemy 会自动追踪变化，生成对应的 UPDATE SQL。

删除（Delete） ​

python
server = session.get(Server, 1)
session.delete(server)
session.commit()

四、会话管理 ​

每个请求应该有自己的会话，请求结束后关闭：

python
from contextlib import contextmanager

@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 使用
with get_db() as db:
    servers = db.execute(select(Server)).scalars().all()

FastAPI 中可以用依赖注入管理会话：

python
from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/servers")
async def list_servers(db: Session = Depends(get_db)):
    return db.execute(select(Server)).scalars().all()

Depends 会在每次请求时调用 get_db()，请求结束后自动关闭会话。

五、常见错误 ​

忘记 commit ​

python
db = SessionLocal()
db.add(Server(hostname="web-01", ip="..."))
# db.commit()  # 忘记提交，数据未保存
db.close()

会话未关闭导致连接泄漏 ​

python
# 错误：异常时不会关闭
db = SessionLocal()
result = db.execute(select(Server)).scalars().all()
db.close()

# 正确：用 try/finally 或上下文管理器
db = SessionLocal()
try:
    result = db.execute(select(Server)).scalars().all()
finally:
    db.close()

在异步函数中用同步 Session ​

python
# 错误：SQLAlchemy 的 Session 是同步的，不能在 async def 中直接 await
@app.get("/servers")
async def list_servers(db: Session = Depends(get_db)):
    return db.execute(select(Server)).scalars().all()   # 同步调用，阻塞事件循环

# 正确 1：路由改为同步
def list_servers(db: Session = Depends(get_db)):
    return db.execute(select(Server)).scalars().all()

# 正确 2：用 SQLAlchemy 的 async 支持（需要 asyncpg/aiosqlite）
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

用旧版 query API 混用新版 select ​

python
# 旧版（SQLAlchemy 1.x）
results = session.query(Server).all()

# 新版（SQLAlchemy 2.0）
results = session.execute(select(Server)).scalars().all()

同一项目应统一用一种风格，推荐新版 select() 语法。