SQLAlchemy 基本概念 #

• Engine：数据库引擎，负责数据库连接和SQL执行。
• Session：会话，负责ORM对象的增删改查和事务管理。
• MetaData：元数据，保存表结构信息。
• Table/Column：表和字段，Core层定义数据库结构。
• ORM模型类：Python类与数据库表的映射。
• Query：ORM查询对象，支持链式调用。

SQLAlchemy 是 Python 中一个功能强大的 ORM（对象关系映射）工具和 SQL 工具包，它提供了灵活的方式来与关系型数据库交互。

1. SQLAlchemy 核心组件 #

SQLAlchemy 主要由两个部分组成：

1. Core：提供 SQL 表达式语言和数据库连接池等功能
2. ORM：建立在 Core 之上的对象关系映射层

2. 安装 SQLAlchemy #

pip install sqlalchemy pymysql

• sqlalchemy是SQLAlchemy的核心库，提供了SQL表达式语言和数据库连接池等功能
• pymysql是Python的MySQL客户端库，用于与MySQL数据库进行交互

常见报错与解决方法

• ModuleNotFoundError: No module named 'pymysql'
  • 解决：pip install pymysql
• sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError)
  • 解决：检查数据库是否启动、用户名密码是否正确、数据库名是否存在。
• sqlalchemy.exc.NoSuchTableError
  • 解决：确认表已创建，或执行Base.metadata.create_all(engine)。

3. 连接数据库 #

# 创建数据库引擎对象，是SQLAlchemy连接数据库的标准方式。
engine = create_engine('dialect+driver://username:password@host:port/database')

• dialect：指定数据库类型，如mysql、postgresql、sqlite等。
• driver：指定数据库驱动，如pymysql、psycopg2等。部分数据库可省略。
• username：数据库用户名。
• password：数据库密码。
• host：数据库服务器地址（本地为localhost）。
• port：数据库端口（MySQL默认3306，PostgreSQL默认5432，SQLite可省略）。
• database：要连接的数据库名称。

示例：

# 从sqlalchemy模块导入create_engine函数
from sqlalchemy import create_engine

# 创建MySQL数据库连接引擎，指定连接字符串
# 格式：mysql+pymysql://用户名:密码@主机地址/数据库名
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')

# 使用try-except语句块来测试数据库连接
try:
    # 通过引擎创建数据库连接对象
    connection = engine.connect()
    # 打印连接成功的提示信息
    print("数据库连接成功！")
    # 关闭数据库连接，释放资源
    connection.close()
except Exception as e:
    # 如果连接失败，捕获异常并打印错误信息
    print(f"数据库连接失败：{e}")

ORM 与 Core 的区别与选择

• Core：直接操作表结构和SQL，适合简单脚本、批量数据处理。
• ORM：面向对象操作数据库，适合复杂业务逻辑和大型项目。
• 建议：新手建议优先学习ORM，理解后再学习Core以应对特殊需求。

4. 使用 Core 组件 #

4.1 定义表结构 #

# 从sqlalchemy模块导入create_engine、MetaData、Table、Column、Integer、String、ForeignKey
from sqlalchemy import create_engine,MetaData, Table, Column, Integer, String, ForeignKey
# 创建MySQL数据库连接引擎，指定连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建元数据对象，用于保存表结构
metadata = MetaData()
# 定义users表，包含id、name、fullname三个字段
users = Table('users', metadata,
    Column('id', Integer, primary_key=True),         # id字段，整型，主键
    Column('name', String(50)),                      # name字段，字符串类型，最大长度50
    Column('fullname', String(50)),                  # fullname字段，字符串类型，最大长度50
)
# 定义addresses表，包含id、user_id、email_address三个字段
addresses = Table('addresses', metadata,
    Column('id', Integer, primary_key=True),                         # id字段，整型，主键
    Column('user_id', None, ForeignKey('users.id')),                 # user_id字段，外键关联users表的id
    Column('email_address', String(100), nullable=False)             # email_address字段，字符串类型，最大长度100，不能为空
)
# 在数据库中创建所有定义的表
metadata.create_all(engine)

真实项目结构推荐

• models.py：定义所有ORM模型类
• database.py：数据库连接和Session管理
• main.py：主程序入口
• crud.py：封装常用的增删改查操作
• alembic/：数据库迁移脚本（推荐使用Alembic管理表结构变更）

示例：

project/
│
├── models.py
├── database.py
├── crud.py
├── main.py
└── alembic/

4.2 执行 SQL 语句 #

# 从sqlalchemy模块导入create_engine、MetaData、Table、Column、Integer、String、ForeignKey
from sqlalchemy import create_engine,MetaData, Table, Column, Integer, String, ForeignKey

# 创建MySQL数据库连接引擎，指定连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')

# 创建元数据对象，用于保存表结构
metadata = MetaData()

# 定义users表，包含id、name、fullname三个字段
users = Table('users', metadata,
    # id字段，整型，主键
    Column('id', Integer, primary_key=True),
    # name字段，字符串类型，最大长度50
    Column('name', String(50)),
    # fullname字段，字符串类型，最大长度50
    Column('fullname', String(50)),
)

# 定义addresses表，包含id、user_id、email_address三个字段
addresses = Table('addresses', metadata,
    # id字段，整型，主键
    Column('id', Integer, primary_key=True),
    # user_id字段，外键关联users表的id
    Column('user_id', None, ForeignKey('users.id')),
    # email_address字段，字符串类型，最大长度100，不能为空
    Column('email_address', String(100), nullable=False)
)

# 在数据库中创建所有定义的表
metadata.create_all(engine)

# 尝试插入数据到users表
try:
    # 使用事务自动提交方式连接数据库
    with engine.begin() as conn:
        # 执行插入操作，插入两条用户数据
        conn.execute(users.insert(), [
            {'name': 'wendy', 'fullname': 'Wendy Williams'},
            {'name': 'mary', 'fullname': 'Mary Contrary'},
        ])
    # 插入成功后输出提示信息
    print("数据插入成功！")
# 捕获异常并输出错误信息
except Exception as e:
    print(f"数据插入失败：{e}")

# 尝试查询users表中的所有数据
try:
    # 连接数据库
    with engine.connect() as conn:
        # 执行查询操作
        result = conn.execute(users.select())
        # 打印查询结果对象
        print(result)
        # 遍历并打印每一行数据
        for row in result:
            print(row)
    # 查询成功后输出提示信息
    print("数据查询成功！")
# 捕获异常并输出错误信息
except Exception as e:
    print(f"数据查询失败：{e}")

5. 使用 ORM 组件 #

5.1 定义模型类 #

# 从sqlalchemy.ext.declarative模块导入declarative_base函数，用于创建基类
from sqlalchemy.ext.declarative import declarative_base
# 从sqlalchemy模块导入create_engine、Column、Integer、String、ForeignKey等必要组件
from sqlalchemy import create_engine,Column, Integer, String, ForeignKey
# 从sqlalchemy.orm模块导入relationship函数，用于定义表间关系
from sqlalchemy.orm import relationship
# 创建MySQL数据库连接引擎，指定连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有模型类都将继承自这个基类
Base = declarative_base()
# 定义User模型类，继承自Base基类
class User(Base):
    # 指定数据库表名为'users'
    __tablename__ = 'users'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，字符串类型
    name = Column(String)
    # 定义fullname字段，字符串类型
    fullname = Column(String)

    # 定义与Address表的一对多关系，back_populates指定反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base基类
class Address(Base):
    # 指定数据库表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，字符串类型，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))

    # 定义与User表的多对一关系，back_populates指定反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有定义的表
Base.metadata.create_all(engine)

5.2 创建会话 #

# 从sqlalchemy.ext.declarative模块导入declarative_base函数，用于创建基类
# 从sqlalchemy模块导入create_engine、Column、Integer、String、ForeignKey等必要组件
from sqlalchemy import create_engine,Column, Integer, String, ForeignKey
# 从sqlalchemy.orm模块导入relationship函数，用于定义表间关系
from sqlalchemy.orm import relationship,declarative_base
# 创建MySQL数据库连接引擎，指定连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有模型类都将继承自这个基类
Base = declarative_base()
# 定义User模型类，继承自Base基类
class User(Base):
    # 指定数据库表名为'users'
    __tablename__ = 'users'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，字符串类型
    name = Column(String)
    # 定义fullname字段，字符串类型
    fullname = Column(String)

    # 定义与Address表的一对多关系，back_populates指定反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base基类
class Address(Base):
    # 指定数据库表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，字符串类型，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))

    # 定义与User表的多对一关系，back_populates指定反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有定义的表
Base.metadata.create_all(engine)

# 从sqlalchemy.orm模块导入sessionmaker函数，用于创建会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

5.3 基本 CRUD 操作 #

创建 (Create) #

# 从sqlalchemy.ext.declarative模块导入declarative_base函数，用于创建基类
# 从sqlalchemy模块导入create_engine、Column、Integer、String、ForeignKey等必要组件
from sqlalchemy import create_engine,Column, Integer, String, ForeignKey
# 从sqlalchemy.orm模块导入relationship函数，用于定义表间关系
from sqlalchemy.orm import relationship,declarative_base
# 创建MySQL数据库连接引擎，指定连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有模型类都将继承自这个基类
Base = declarative_base()
# 定义User模型类，继承自Base基类
class User(Base):
    # 指定数据库表名为'users'
    __tablename__ = 'users'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，字符串类型
    name = Column(String)
    # 定义fullname字段，字符串类型
    fullname = Column(String)

    # 定义与Address表的一对多关系，back_populates指定反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base基类
class Address(Base):
    # 指定数据库表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，字符串类型，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))

    # 定义与User表的多对一关系，back_populates指定反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有定义的表
Base.metadata.create_all(engine)

# 从sqlalchemy.orm模块导入sessionmaker函数，用于创建会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

+# 创建新用户对象
+new_user = User(name='ed', fullname='Ed Jones')
+# 将新用户对象添加到会话
+session.add(new_user)
+# 提交会话，将数据写入数据库
+session.commit()

查询 (Read) #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

# 创建一个新的User对象
new_user = User(name='ed', fullname='Ed Jones')
# 将新用户对象添加到会话中
session.add(new_user)
# 提交会话，将新用户写入数据库
session.commit()

+# 查询所有User表中的用户
+users = session.query(User).all()
+# 遍历查询结果并打印每个用户的fullname
+for user in users:
+    print(user.fullname)
+# 按条件查询name为'ed'的用户
+user = session.query(User).filter_by(name='ed').first()
+# 打印该用户的fullname
+print(user.fullname)
+# 导入or_用于复杂查询
+from sqlalchemy import or_
+# 查询name为'ed'或'wendy'的用户
+results = session.query(User).filter(
+    or_(User.name == 'ed', User.name == 'wendy')
+).all()
+# 遍历查询结果并打印每个用户的fullname
+for result in results:
+    print(result.fullname)

更新 (Update) #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

# 创建一个新的User对象
new_user = User(name='ed', fullname='Ed Jones')
# 将新用户对象添加到会话中
session.add(new_user)
# 提交会话，将新用户写入数据库
session.commit()

# 查询所有User表中的用户
users = session.query(User).all()
# 遍历查询结果并打印每个用户的fullname
for user in users:
    print(user.fullname)
# 按条件查询name为'ed'的用户
user = session.query(User).filter_by(name='ed').first()
# 打印该用户的fullname
print(user.fullname)
# 导入or_用于复杂查询
from sqlalchemy import or_
# 查询name为'ed'或'wendy'的用户
results = session.query(User).filter(
    or_(User.name == 'ed', User.name == 'wendy')
).all()
# 遍历查询结果并打印每个用户的fullname
for result in results:
    print(result.fullname)
# 查询name为'ed'的用户对象
user = session.query(User).filter_by(name='ed').first()
# 修改该用户的fullname字段
user.fullname = 'Edward Jones'
# 提交会话，将更改保存到数据库
session.commit()

删除 (Delete) #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

# 创建一个新的User对象
new_user = User(name='ed', fullname='Ed Jones')
# 将新用户对象添加到会话中
session.add(new_user)
# 提交会话，将新用户写入数据库
session.commit()

# 查询所有User表中的用户
users = session.query(User).all()
# 遍历查询结果并打印每个用户的fullname
for user in users:
    print(user.fullname)
# 按条件查询name为'ed'的用户
user = session.query(User).filter_by(name='ed').first()
# 打印该用户的fullname
print(user.fullname)
# 导入or_用于复杂查询
from sqlalchemy import or_
# 查询name为'ed'或'wendy'的用户
results = session.query(User).filter(
    or_(User.name == 'ed', User.name == 'wendy')
).all()
# 遍历查询结果并打印每个用户的fullname
for result in results:
    print(result.fullname)
# 查询name为'ed'的用户对象
user = session.query(User).filter_by(name='ed').first()
# 修改该用户的fullname字段
user.fullname = 'Edward Jones'
# 提交会话，将更改保存到数据库
session.commit()
# 查询name为'ed'的用户对象
user = session.query(User).filter_by(name='ed').first()
# 删除该用户对象
session.delete(user)
# 提交会话，将删除操作保存到数据库
session.commit()

易错点与小贴士

• 插入/更新/删除后要记得session.commit()，否则数据不会写入数据库。
• 查询返回的是对象列表，遍历时注意属性名。
• 查询单条数据用.first()，多条用.all()。
• 复杂查询建议多用filter、filter_by、or_、and_等组合。
• 事务出错时要session.rollback()。

6. 高级特性 #

6.1 关系加载策略 #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()

# 创建一个新的User对象
new_user = User(name='ed', fullname='Ed Jones')
# 将新用户对象添加到会话中
session.add(new_user)
# 提交会话，将新用户写入数据库
session.commit()

# 查询所有User表中的用户
users = session.query(User).all()
# 遍历查询结果并打印每个用户的fullname
for user in users:
    print(user.fullname)
# 按条件查询name为'ed'的用户
user = session.query(User).filter_by(name='ed').first()
# 打印该用户的fullname
print(user.fullname)
# 导入or_用于复杂查询
from sqlalchemy import or_
# 查询name为'ed'或'wendy'的用户
results = session.query(User).filter(
    or_(User.name == 'ed', User.name == 'wendy')
).all()
# 遍历查询结果并打印每个用户的fullname
for result in results:
    print(result.fullname)
# 查询name为'ed'的用户对象
user = session.query(User).filter_by(name='ed').first()
# 修改该用户的fullname字段
user.fullname = 'Edward Jones'
# 提交会话，将更改保存到数据库
session.commit()
# 查询name为'ed'的用户对象
user = session.query(User).filter_by(name='ed').first()
# 删除该用户对象
session.delete(user)
# 提交会话，将删除操作保存到数据库
session.commit()

# 创建一个新的Address对象
new_address = Address(email_address='ed@example.com',user_id=9)
# 将新地址对象添加到会话中
session.add(new_address)
# 提交会话，将新地址写入数据库
session.commit()

# 导入joinedload用于实现立即加载（Eager Loading）
from sqlalchemy.orm import joinedload
# 查询所有User对象，并立即加载其关联的addresses数据
users = session.query(User).options(joinedload(User.addresses)).all()
# 遍历所有用户
for user in users:
    # 打印用户的fullname
    print(user.fullname)
    # 遍历该用户的所有地址
    for address in user.addresses:
        # 打印地址的email_address
        print(address.email_address)

# 查询第一个User对象，采用延迟加载（Lazy Loading，默认方式）
user = session.query(User).first()
# 访问user.addresses属性时，才会从数据库加载关联的addresses数据
addresses = user.addresses  # 此时才会加载关联数据
# 遍历该用户的所有地址
for address in addresses:
    # 打印地址的email_address
    print(address.email_address)

6.2 事务管理 #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建会话对象
session = Session()
# 使用try-except结构进行事务管理，确保数据一致性
try:
    # 创建一个名为'user1'的新用户对象，fullname为'User One'
    user1 = User(name='user1', fullname='User One')
    # 将user1对象添加到会话中，准备写入数据库
    session.add(user1)

    # 创建另一个名为'user2'的新用户对象，fullname为'User Two'
    user2 = User(name='user2', fullname='User Two')
    # 将user2对象添加到会话中
    session.add(user2)

    # 提交会话，将所有更改保存到数据库
    session.commit()
# 如果发生异常，则回滚事务，撤销所有未提交的更改
except:
    session.rollback()
    # 重新抛出异常，便于上层捕获或调试
    raise

6.3 混合属性 (Hybrid Attributes) #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")
    # 使用hybrid_property装饰器定义混合属性，该属性既可以在Python对象上使用，也可以在SQL查询中使用
    @hybrid_property
    # 定义description方法，用于生成用户的描述信息
    def description(self):
        # 使用列表推导式获取用户所有地址的邮箱地址，并用逗号连接成字符串
        email_addresses = ', '.join([addr.email_address for addr in self.addresses])
        # 返回包含用户全名和邮箱地址的描述字符串
        return f"{self.fullname} {email_addresses}"

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建数据库会话对象，用于与数据库进行交互
session = Session()
# 查询User表中的第一个用户对象
user = session.query(User).first()
# 打印用户的描述信息（包含姓名和邮箱地址）
print(user.description)

进阶阅读与官方文档链接

• SQLAlchemy 官方文档（中文）
• SQLAlchemy ORM 官方教程
• SQLAlchemy Core 官方教程
• Alembic 官方文档（数据库迁移）
• SQLAlchemy 常见问题解答

7. 性能优化 #

1. 批量插入：使用 bulk_insert_mappings
2. 批量更新：使用 bulk_update_mappings
3. 只查询需要的列：避免 SELECT *
4. 使用索引：在频繁查询的列上创建索引
5. 合理使用连接加载策略

7.1 bulk_insert_mappings #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")
    # 使用hybrid_property装饰器定义混合属性，该属性既可以在Python对象上使用，也可以在SQL查询中使用
    @hybrid_property
    # 定义description方法，用于生成用户的描述信息
    def description(self):
        # 使用列表推导式获取用户所有地址的邮箱地址，并用逗号连接成字符串
        email_addresses = ', '.join([addr.email_address for addr in self.addresses])
        # 返回包含用户全名和邮箱地址的描述字符串
        return f"{self.fullname} {email_addresses}"

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建数据库会话对象，用于与数据库进行交互
session = Session()
# 定义要批量插入User表的数据，每个字典代表一条用户记录
users_data = [
    {"name": "alice", "fullname": "Alice Wonderland"},
    {"name": "bob", "fullname": "Bob Builder"},
    {"name": "cathy", "fullname": "Cathy Smith"}
]
# 使用bulk_insert_mappings方法将users_data批量插入到User表中
session.bulk_insert_mappings(User, users_data)
# 提交事务，将插入操作保存到数据库
session.commit()
# 查询User表中的所有用户，并打印每个用户的姓名和用户名
for user in session.query(User).all():
    print(f"姓名: {user.fullname}，用户名: {user.name}")

7.2 bulk_update_mappings #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")
    # 使用hybrid_property装饰器定义混合属性，该属性既可以在Python对象上使用，也可以在SQL查询中使用
    @hybrid_property
    # 定义description方法，用于生成用户的描述信息
    def description(self):
        # 使用列表推导式获取用户所有地址的邮箱地址，并用逗号连接成字符串
        email_addresses = ', '.join([addr.email_address for addr in self.addresses])
        # 返回包含用户全名和邮箱地址的描述字符串
        return f"{self.fullname} {email_addresses}"

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建数据库会话对象，用于与数据库进行交互
session = Session()
# 定义要批量更新User表的数据，每个字典包含要更新的主键id和新字段值
update_data = [
    {"id": 9, "fullname": "Alice Liddell"},
    {"id": 10, "fullname": "Bob the Builder"},
    {"id": 15, "fullname": "Catherine Smith"}
]
# 使用bulk_update_mappings方法批量更新User表
session.bulk_update_mappings(User, update_data)
# 提交事务，将更新操作保存到数据库
session.commit()
# 再次查询User表，打印更新后的用户信息
print("批量更新后：")
for user in session.query(User).all():
    print(f"姓名: {user.fullname}，用户名: {user.name}")

8. 常见使用模式 #

8.1 分页查询 #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")
    # 使用hybrid_property装饰器定义混合属性，该属性既可以在Python对象上使用，也可以在SQL查询中使用
    @hybrid_property
    # 定义description方法，用于生成用户的描述信息
    def description(self):
        # 使用列表推导式获取用户所有地址的邮箱地址，并用逗号连接成字符串
        email_addresses = ', '.join([addr.email_address for addr in self.addresses])
        # 返回包含用户全名和邮箱地址的描述字符串
        return f"{self.fullname} {email_addresses}"

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建数据库会话对象，用于与数据库进行交互
session = Session()

# 设置当前页码为1
page = 1
# 设置每页显示的用户数量为5
per_page = 5
# 查询User表，按id升序排序，跳过前(page-1)*per_page条，取per_page条数据，实现分页
users = session.query(User).order_by(User.id).offset((page-1)*per_page).limit(per_page).all()
# 遍历查询到的用户列表
for user in users:
    # 打印每个用户的fullname字段
    print(user.fullname)

8.2 聚合查询 #

# 导入SQLAlchemy的create_engine、Column、Integer、String、ForeignKey等组件
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
# 导入relationship用于定义表间关系，declarative_base用于创建ORM基类
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
# 创建MySQL数据库连接引擎，指定数据库连接字符串
engine = create_engine('mysql+pymysql://root:123456@localhost/jobs')
# 创建声明性基类，所有ORM模型类都将继承自该基类
Base = declarative_base()

# 定义User模型类，继承自Base
class User(Base):
    # 指定该模型对应的表名为'users'
    __tablename__ = 'users'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义name字段，类型为字符串
    name = Column(String)
    # 定义fullname字段，类型为字符串
    fullname = Column(String)
    # 定义与Address表的一对多关系，back_populates用于反向引用
    addresses = relationship("Address", back_populates="user")
    # 使用hybrid_property装饰器定义混合属性，该属性既可以在Python对象上使用，也可以在SQL查询中使用
    @hybrid_property
    # 定义description方法，用于生成用户的描述信息
    def description(self):
        # 使用列表推导式获取用户所有地址的邮箱地址，并用逗号连接成字符串
        email_addresses = ', '.join([addr.email_address for addr in self.addresses])
        # 返回包含用户全名和邮箱地址的描述字符串
        return f"{self.fullname} {email_addresses}"

# 定义Address模型类，继承自Base
class Address(Base):
    # 指定该模型对应的表名为'addresses'
    __tablename__ = 'addresses'

    # 定义id字段，类型为整型，主键
    id = Column(Integer, primary_key=True)
    # 定义email_address字段，类型为字符串，不能为空
    email_address = Column(String, nullable=False)
    # 定义user_id字段，类型为整型，外键关联users表的id字段
    user_id = Column(Integer, ForeignKey('users.id'))
    # 定义与User表的多对一关系，back_populates用于反向引用
    user = relationship("User", back_populates="addresses")

# 在数据库中创建所有已定义的表
Base.metadata.create_all(engine)

# 导入sessionmaker用于创建数据库会话
from sqlalchemy.orm import sessionmaker
# 创建会话工厂，并绑定到数据库引擎
Session = sessionmaker(bind=engine)
# 创建数据库会话对象，用于与数据库进行交互
session = Session()

# 导入SQLAlchemy的func模块，用于调用SQL聚合函数
from sqlalchemy import func

# 查询User表中所有用户的数量，返回总数
count = session.query(func.count(User.id)).scalar()
# 打印用户总数
print(count)

# 查询每个用户的name以及其对应的地址数量
result = session.query(User.name, func.count(Address.id)).join(Address).group_by(User.name).all()
# 打印每个用户的姓名及其地址数量
print(result)

9. 最佳实践 #

1. 分离模型定义：将模型类放在单独的模块中
2. 使用会话管理：确保会话正确关闭
3. 避免长事务：尽快提交或回滚事务
4. 合理设计关系：避免N+1查询问题
5. 使用Alembic进行数据库迁移：管理数据库模式变更

10. 总结 #

SQLAlchemy 提供了从低级 SQL 操作到高级 ORM 功能的完整解决方案。它的灵活性允许开发者根据项目需求选择合适的使用方式，从小型应用到大型企业系统都能胜任。掌握 SQLAlchemy 可以显著提高 Python 数据库应用的开发效率和质量。