SQLAlchemy ORM 核心机制解析

SQLAlchemy 是 Python 中功能强大的对象关系映射（ORM）工具，它将数据库操作抽象为面向对象的编程方式。通过 SQLAlchemy，开发者可以使用类和实例来表示数据库表和记录，从而避免直接编写复杂的 SQL 语句。

数据库连接配置

SQLAlchemy 本身不直接操作数据库，而是依赖于底层的数据库驱动程序，如 PyMySQL、mysqlclient 等。它通过 Dialect 组件与这些驱动交互，实现对不同数据库的统一访问。以下是常见的数据库连接字符串格式：

MySQL (PyMySQL): mysql+pymysql://user:pass@host:port/dbname
PostgreSQL: postgresql+psycopg2://user:pass@host:port/dbname
SQLite: sqlite:///path/to/database.db

引擎与会话初始化

创建数据库连接的核心是 create_engine 函数，它返回一个 Engine 实例，负责管理连接池和执行 SQL 命令。会话（Session）则用于跟踪对象的状态变更。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建数据库引擎
engine = create_engine(
    "mysql+pymysql://root:password@localhost:3306/myapp",
    pool_size=5,
    max_overflow=10,
    pool_recycle=3600
)

# 创建基类和会话工厂
Base = declarative_base()
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()

模型定义与表结构创建

通过继承 declarative_base() 返回的基类，可以定义数据模型。每个模型类对应一张数据库表，类属性对应表的列。

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)

class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(String(4000))
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)

定义完模型后，调用 Base.metadata.create_all(engine) 即可自动在数据库中创建对应的表结构。

基本数据操作

插入数据

# 添加单个对象
new_user = User(username='alice', email='alice@example.com', age=28)
session.add(new_user)

# 批量添加
session.add_all([
    User(username='bob', email='bob@example.com'),
    User(username='charlie', email='charlie@example.com')
])

session.commit()  # 提交事务

查询数据

SQLAlchemy 提供了灵活的查询接口，支持多种过滤和排序方式。

# 基本查询
all_users = session.query(User).all()
user_by_id = session.get(User, 1)  # 根据主键获取

# 条件过滤
active_users = session.query(User)\
    .filter(User.age >= 18)\
    .filter(User.username.like('a%'))\
    .all()

# 复合条件
from sqlalchemy import or_
results = session.query(User)\
    .filter(or_(User.username == 'admin', User.email.contains('@company.com')))\
    .first()

# 排序与分页
ordered = session.query(User)\
    .order_by(User.username.desc())\
    .limit(10)\
    .offset(0)\
    .all()

更新数据

# 批量更新
session.query(User)\
    .filter(User.age < 18)\
    .update({'age': 18}, synchronize_session=False)

# 更新单个对象
target_user = session.query(User).filter_by(username='alice').first()
if target_user:
    target_user.email = 'new_email@example.com'
session.commit()

删除数据

# 删除指定记录
session.query(User)\
    .filter(User.username == 'temp_user')\
    .delete(synchronize_session=False)

# 删除对象实例
user_to_remove = session.query(User).get(999)
if user_to_remove:
    session.delete(user_to_remove)
session.commit()

关联关系处理

SQLAlchemy 支持多种关系模式，包括一对多、多对一和多对多。

from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 一对多：一个用户拥有多个文章
    articles = relationship("Article", back_populates="author", cascade="all, delete-orphan")

class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey('users.id'))
    
    # 多对一：一篇文章属于一个作者
    author = relationship("User", back_populates="articles")

多对多关系示例

from sqlalchemy import Table

# 关联表
user_role_table = Table(
    'user_roles', Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('role_id', Integer, ForeignKey('roles.id'))
)

class Role(Base):
    __tablename__ = 'roles'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    # 多对多关系
    roles = relationship("Role", secondary=user_role_table, backref="users")

高级查询技巧

聚合函数

from sqlalchemy import func

# 统计用户数量
count = session.query(func.count(User.id)).scalar()

# 计算平均年龄
avg_age = session.query(func.avg(User.age)).scalar()

# 分组统计
stats = session.query(
    func.min(User.age),
    func.max(User.age),
    func.avg(User.age)
).first()

连接查询

# 内连接
result = session.query(User, Article)\
    .join(Article, User.id == Article.author_id)\
    .filter(Article.title.contains('Python'))\
    .all()

# 使用 relationship 进行连接
result = session.query(User)\
    .join(User.articles)\
    .filter(Article.title == 'Introduction to SQLAlchemy')\
    .all()

事务控制

SQLAlchemy 支持完整的事务管理，确保数据操作的原子性。

try:
    # 开始事务
    new_user = User(username='transaction_test', email='test@example.com')
    session.add(new_user)
    
    # 模拟业务逻辑
    if some_condition_fails:
        raise Exception("Simulated failure")
    
    session.commit()  # 提交事务
except Exception as e:
    session.rollback()  # 回滚事务
    print(f"Transaction failed: {e}")
finally:
    session.close()