SQLAlchemy 详细教程

SQLAlchemy 是 Python 中最强大且灵活的 ORM（对象关系映射）工具之一，它提供了完整的 SQL 功能和高层的 ORM 抽象。本教程将带你从基础到高级使用 SQLAlchemy。

1. 安装与配置

pip install sqlalchemy

对于特定数据库还需要安装对应的驱动，例如：

# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# SQLite (Python内置支持，无需额外安装)

2. 核心概念

Engine：数据库连接的起点，管理连接池
Session：与数据库交互的主要接口
Model：映射到数据库表的Python类
Query：用于构建和执行数据库查询

3. 定义模型

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    fullname = Column(String(50))
    nickname = Column(String(50))

    # 定义一对多关系
    addresses = relationship("Address", back_populates="user")
class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True)
    email_address = Column(String(100), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    # 定义多对一关系
    user = relationship("User", back_populates="addresses")

4. 创建会话

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 创建引擎 (这里使用SQLite内存数据库)
engine = create_engine('sqlite:///:memory:', echo=True)
# 创建所有表
Base.metadata.create_all(engine)
# 创建会话工厂
Session = sessionmaker(bind=engine)
# 创建会话实例
session = Session()

5. 基本CRUD操作

创建(Create)

# 创建新用户
new_user = User(name='john', fullname='John Doe', nickname='johny')
session.add(new_user)
# 批量添加
session.add_all([
    User(name='wendy', fullname='Wendy Williams', nickname='windy'),
    User(name='mary', fullname='Mary Contrary', nickname='mary'),
    User(name='fred', fullname='Fred Flintstone', nickname='freddy')
])
# 提交到数据库
session.commit()

读取(Read)

# 获取单个用户
user = session.query(User).filter_by(name='john').first()
# 获取所有用户
users = session.query(User).all()

更新(Update)

# 修改用户属性
user.nickname = 'johnny'
# 直接更新
session.query(User).filter_by(name='john').update({'nickname': 'johny'})
session.commit()

删除(Delete)

# 删除用户
user = session.query(User).filter_by(name='fred').first()
session.delete(user)
# 直接删除
session.query(User).filter_by(name='mary').delete()
session.commit()

6. 查询数据

基本查询

# 获取所有用户
users = session.query(User).all()
# 获取第一个用户
first_user = session.query(User).first()
# 获取数量
user_count = session.query(User).count()
# 限制和偏移
users = session.query(User).limit(10).offset(5)

过滤查询

from sqlalchemy import or_
# 简单过滤
john = session.query(User).filter(User.name == 'john').first()
# 多条件过滤
users = session.query(User).filter(
    User.name.like('%j%'),
    User.fullname != 'John Doe'
).all()
# 或条件
users = session.query(User).filter(
    or_(
        User.name == 'john',
        User.name == 'wendy'
    )
).all()

排序

# 简单排序
users = session.query(User).order_by(User.name).all()
# 多列排序
users = session.query(User).order_by(User.name.desc(), User.id).all()

7. 关系与连接

一对多关系

# 添加地址
john = session.query(User).filter_by(name='john').first()
john.addresses = [
    Address(email_address='john@example.com'),
    Address(email_address='john@gmail.com')
]
session.commit()
# 查询关联数据
addresses = session.query(Address).join(Address.user).filter(User.name == 'john').all()
# 或者通过关系属性
for addr in john.addresses:
    print(addr.email_address)

多对多关系

from sqlalchemy import Table, Text
# 关联表
post_keywords = Table('post_keywords', Base.metadata,
    Column('post_id', ForeignKey('posts.id'), primary_key=True),
    Column('keyword_id', ForeignKey('keywords.id'), primary_key=True)
)
class BlogPost(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    headline = Column(String(255), nullable=False)
    body = Column(Text)

    # 多对多关系
    keywords = relationship("Keyword", secondary=post_keywords, back_populates="posts")

    # 多对一关系
    author = relationship("User", back_populates="posts")
class Keyword(Base):
    __tablename__ = 'keywords'

    id = Column(Integer, primary_key=True)
    keyword = Column(String(50), nullable=False, unique=True)

    posts = relationship("BlogPost", secondary=post_keywords, back_populates="keywords")
# 添加到User模型
User.posts = relationship("BlogPost", back_populates="author", lazy="dynamic")

8. 事务管理

try:
    # 开始事务
    user1 = User(name='user1', fullname='User One', nickname='u1')
    session.add(user1)

    # 这里可以执行更多操作

    # 提交事务
    session.commit()
except:
    # 发生错误时回滚
    session.rollback()
    raise
finally:
    # 关闭会话
    session.close()

9. 高级查询

聚合函数

from sqlalchemy import func
# 计数
count = session.query(func.count(User.id)).scalar()
# 分组统计
result = session.query(
    User.name,
    func.count(Address.id).label('address_count')
).join(Address).group_by(User.name).all()

子查询

from sqlalchemy.sql import label
subq = session.query(
    Address.user_id,
    label('address_count', func.count(Address.id))
).group_by(Address.user_id).subquery()
users = session.query(
    User.name,
    User.fullname,
    subq.c.address_count
).join(subq, User.id == subq.c.user_id).all()

原生SQL

# 执行原生SQL
result = session.execute("SELECT * FROM users WHERE name=:name", {'name': 'john'})
for row in result:
    print(row)

10. 数据库迁移

SQLAlchemy本身不提供迁移工具，但可以与Alembic配合使用：

pip install alembic

初始化Alembic：

alembic init alembic

配置alembic.ini和alembic/env.py后，可以创建迁移：

alembic revision --autogenerate -m "Added user table"
alembic upgrade head

11. 性能优化

延迟加载 vs 立即加载

# 默认是延迟加载
user = session.query(User).first()
print(user.addresses)  # 这里会触发额外查询
# 使用joinedload立即加载关联数据
from sqlalchemy.orm import joinedload
user = session.query(User).options(joinedload(User.addresses)).first()
print(user.addresses)  # 不会触发额外查询

批量操作

# 批量插入
session.bulk_save_objects([
    User(name="u1"),
    User(name="u2"),
    User(name="u3")
])
# 批量更新
session.query(User).filter(User.name.like("u%")).update(
    {"fullname": "Updated Name"},
    synchronize_session=False
)

缓存查询

from sqlalchemy.orm import with_polymorphic
# 使用缓存查询
query = session.query(User).options(
    with_polymorphic([User], [Address])
)

总结

SQLAlchemy提供了强大而灵活的数据库访问能力，从简单的CRUD操作到复杂的关系处理和高级查询。通过合理使用其特性，可以构建高效、可维护的数据库访问层。要深入学习SQLAlchemy，建议：

阅读官方文档
实践各种查询和关系模式
学习性能优化技巧
结合Alembic掌握数据库迁移希望本教程能帮助你快速掌握SQLAlchemy的核心功能！

SQLAlchemy 详细教程

SQLAlchemy 详细教程

目录

1. 安装与配置

2. 核心概念

3. 定义模型

4. 创建会话

5. 基本CRUD操作

创建(Create)

读取(Read)

更新(Update)

删除(Delete)

6. 查询数据

基本查询

过滤查询

排序

7. 关系与连接

一对多关系

多对多关系

8. 事务管理

9. 高级查询

聚合函数

子查询

原生SQL

10. 数据库迁移

11. 性能优化

延迟加载 vs 立即加载

批量操作

缓存查询

总结

results matching ""

No results matching ""