Python SQLAlchemy ORM 实战

发布于 约 13 分钟

直接写 SQL 容易出错且难维护，SQLAlchemy 把数据库表映射成 Python 类，用面向对象的方式操作数据库。本文从连接、模型到查询和关联，讲清 SQLAlchemy ORM 的核心用法。

SQLAlchemy 配图：蓝色渐变背景上数据库圆柱通过发光管线连接中央六边形枢纽与节点，扁平科技插画
SQLAlchemy：用 Python 对象操作数据库

一、为什么用 ORM

直接写 SQL 有几个问题：字符串拼接容易引入 SQL 注入风险、表结构改了要到处改 SQL、调试不方便。ORM（对象关系映射）把数据库表映射成 Python 类，用操作对象的方式操作数据库。

# 不用 ORM：手写 SQL
import sqlite3
conn = sqlite3.connect("test.db")
cursor = conn.execute("SELECT * FROM users WHERE age > ?", (18,))
users = cursor.fetchall()

# 用 ORM：操作对象
from models import User
adults = User.query.filter(User.age > 18).all()
ORM 选型：Python 生态最主流的是 SQLAlchemy（成熟稳定）和 Peewee（轻量简单）。新项目推荐 SQLAlchemy 2.0，支持类型提示和现代写法。

二、安装与连接

pip install sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# 创建引擎（连接池）
engine = create_engine(
    "sqlite:///test.db",  # 数据库 URL
    echo=True,            # 打印 SQL（开发时用）
    pool_size=5,          # 连接池大小
)

# 基类
class Base(DeclarativeBase):
    pass

# Session 工厂
SessionLocal = sessionmaker(bind=engine)

三、定义模型

from sqlalchemy import String, Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

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(100), unique=True)
    age: Mapped[int] = mapped_column(Integer, default=0)

    # 关联：一个用户有多篇文章
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped[User] = relationship(back_populates="posts")

# 创建所有表
Base.metadata.create_all(engine)

四、增删改查

db = SessionLocal()

# 增
user = User(name="小明", email="xm@example.com", age=25)
db.add(user)
db.commit()

# 查
user = db.get(User, 1)  # 按主键查
users = db.query(User).filter(User.age > 18).all()  # 条件查
first = db.query(User).filter(User.name == "小明").first()

# 改
user = db.get(User, 1)
user.age = 26
db.commit()

# 删
db.delete(user)
db.commit()

db.close()

五、关联查询

# 查询用户的所有文章
user = db.get(User, 1)
for post in user.posts:
    print(post.title)

# 查询文章的作者
post = db.get(Post, 1)
print(post.author.name)

# 联表查询：查所有文章及其作者名
from sqlalchemy.orm import joinedload
posts = db.query(Post).options(joinedload(Post.author)).all()
for p in posts:
    print(p.title, p.author.name)

六、高级查询

from sqlalchemy import func

# 聚合：按年龄分组统计人数
result = db.query(User.age, func.count(User.id)).group_by(User.age).all()

# 排序 + 分页
page = db.query(User).order_by(User.id.desc()).offset(0).limit(10).all()

# 模糊查询
search = db.query(User).filter(User.name.like("%明%")).all()

# 原生 SQL（复杂查询时）
from sqlalchemy import text
result = db.execute(text("SELECT * FROM users WHERE age > :age"), {"age": 18}).all()

七、总结

SQLAlchemy ORM 让数据库操作变成面向对象的方式：定义模型类代替手写建表 SQL，用 query API 代替字符串拼接查询。关联查询通过 relationship 自动处理 JOIN。记住：简单查询用 ORM，复杂查询可以混合原生 SQL，两者不冲突。

延伸阅读