sqlalchemy

1. SQLAlchemy 是什么？

SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）框架 。简单来说，它让你用 Python 类和对象来操作数据库，而不用直接写 SQL。

1.1 为什么需要 SQLAlchemy？

如果不用 ORM，直接操作数据库：

# 传统方式：直接写 SQL（容易出错，代码冗长）
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("张三", 25))
cursor.execute("SELECT * FROM users WHERE age > 25")

使用 SQLAlchemy 推荐方式：

# ORM方式：使用Python对象，代码更安全直观
user = User(name="张三", age=25)
session.add(user)
session.commit()

users = session.scalars(select(User).where(User.age > 25)).all()

1.2 SQLAlchemy 的核心优势

代码简洁直观 ：用对象而不是SQL字符串操作数据
类型安全 ：更早发现问题
数据库无关 ：可无缝切换不同数据库
自动管理连接与事务 ：无需手动管理细节

2. 前置知识

学习 SQLAlchemy 之前，需要了解：

2.1 什么是 ORM？

ORM（Object-Relational Mapping，对象关系映射） ：把数据库表映射为 Python 类，行映射为对象，列映射为对象属性。

比如：

表 users → Python类 User
一行数据 → User实例
列（如 name、age）→ 对象属性

2.2 数据库基础术语

表（Table） ：存储结构，类似 Excel 表
行（Row） ：一条记录
列（Column） ：字段
主键（Primary Key） ：唯一标识
外键（Foreign Key） ：关联其他表

2.3 Python 基础

类及实例的定义与使用
上下文管理器：with语句

完全不熟悉可先补基础。

3. 安装与环境准备

3.1 安装 SQLAlchemy

使用pip：

pip install sqlalchemy

3.2 选择数据库

本教程用SQLite 示例，因为：

免服务器、易上手
Python自带支持
学习/开发首选

其它数据库（如MySQL、PostgreSQL）需安装驱动：

MySQL: pip install pymysql
PostgreSQL: pip install psycopg2

4. 核心概念：Engine、Base、Session

使用 SQLAlchemy 2.0+，主要概念如下：

4.1 Engine（引擎）

Engine 负责数据库连接和方言处理。

from sqlalchemy import create_engine

engine = create_engine("sqlite:///example.db", echo=True)

SQLite: sqlite:///文件名.db
MySQL: mysql+pymysql://用户名:密码@主机:端口/数据库名
PostgreSQL: postgresql://用户名:密码@主机:端口/数据库名

4.2 Base（声明式基类）

Base 是所有ORM模型的基类。2.0推荐定义如下：

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

4.3 Session（会话）

Session 负责ORM操作，2.0+ 推荐用 with 上下文：

from sqlalchemy.orm import Session

with Session(engine) as session:
    # 数据库操作
    ...

5. 第一个完整示例：从零开始

下面是SQLAlchemy 2.0+ 推荐风格的完整脚本：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column
from sqlalchemy import select

# 定义Base
class Base(DeclarativeBase):
    pass

# 定义User模型
class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

# 创建引擎
engine = create_engine("sqlite:///first_example.db", echo=True)
# 创建表
Base.metadata.create_all(engine)

# 用Session上下文增加和查询数据
with Session(engine) as session:
    new_user = User(name="张三", age=25)
    session.add(new_user)
    session.commit()

    users = session.scalars(select(User)).all()
    print("所有用户：")
    for user in users:
        print(user)

使用方法 ：

保存为 first_example.py
执行 python first_example.py
查看控制台SQL输出和结果
目录出现 first_example.db

6. 定义数据模型

模型即Python类，继承自 Base，推荐如下：

6.1 基本模型定义

from sqlalchemy import create_engine, String, Integer, Float, Boolean, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    price: Mapped[float]
    is_available: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)

    def __repr__(self):
        return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"

engine = create_engine("sqlite:///model_example.db", echo=True)
Base.metadata.create_all(engine)
print("表创建成功！")

6.2 常用列类型

Integer ：整数
String(n) ：字符串（n最大长度）
Float ：浮点
Boolean ：布尔
DateTime ：日期时间
Text ：长文本
Date ：日期

6.3 列参数

primary_key=True ：主键
nullable=False ：非空
default=值 ：默认值
unique=True ：唯一

7. 基本 CRUD 操作

CRUD：Create（增） 、Read（查） 、Update（改） 、Delete（删）

7.1 创建（Create）：添加数据

from sqlalchemy import create_engine, String, Integer
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///crud_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 单个添加
    user1 = User(name="张三", age=25)
    session.add(user1)
    session.commit()
    print("添加用户1成功")

    # 批量添加
    users = [
        User(name="李四", age=30),
        User(name="王五", age=28),
        User(name="赵六", age=35),
    ]
    session.add_all(users)
    session.commit()
    print("批量添加用户成功")

7.2 读取（Read）：查询数据

from sqlalchemy import create_engine, String, Integer, select, desc
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///crud_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 查询所有
    all_users = session.scalars(select(User)).all()
    print("所有用户：")
    for user in all_users:
        print(user)

    # 查询第一条
    first_user = session.scalars(select(User)).first()
    print(f"\n第一条记录：{first_user}")

    # 根据主键查
    user_by_id = session.get(User, 1)
    print(f"\nID为1的用户：{user_by_id}")

    # filter_by等值查询
    user_zhang = session.scalars(select(User).where(User.name == "张三")).first()
    print(f"\n姓名为'张三'的用户：{user_zhang}")

    # filter复杂条件
    young_users = session.scalars(select(User).where(User.age < 30)).all()
    print(f"\n年龄小于30的用户：")
    for user in young_users:
        print(user)

    # 排序
    sorted_users = session.scalars(select(User).order_by(desc(User.age))).all()
    print(f"\n按年龄降序排列：")
    for user in sorted_users:
        print(user)

    # 限制数量
    top_2 = session.scalars(select(User).order_by(desc(User.age)).limit(2)).all()
    print(f"\n年龄最大的2个用户：")
    for user in top_2:
        print(user)

    # 统计
    user_count = session.execute(select(User).count()).scalar_one()
    print(f"\n用户总数：{user_count}")

7.3 更新（Update）：修改数据

from sqlalchemy import create_engine, String, Integer, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///crud_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 先查后改
    user = session.scalars(select(User).where(User.name == "张三")).first()
    if user:
        user.age = 26
        session.commit()
        print(f"更新成功：{user}")
    else:
        print("用户不存在")

    # 批量更新
    session.execute(
        select(User).where(User.age > 30).execution_options(synchronize_session="fetch")
    ).scalars().update({"age": 31})
    session.commit()
    print("批量更新成功")

    # 查询更新后
    updated_users = session.scalars(select(User)).all()
    print("\n更新后的所有用户：")
    for user in updated_users:
        print(user)

7.4 删除（Delete）：删除数据

from sqlalchemy import create_engine, String, Integer, select, delete
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///crud_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 删除单个
    user = session.scalars(select(User).where(User.name == "张三")).first()
    if user:
        session.delete(user)
        session.commit()
        print("删除成功")
    else:
        print("用户不存在")

    # 批量删除
    deleted_count = session.execute(
        delete(User).where(User.age < 25)
    ).rowcount
    session.commit()
    print(f"批量删除了{deleted_count}条记录")

    # 查询剩余
    remaining_users = session.scalars(select(User)).all()
    print("\n剩余用户：")
    for user in remaining_users:
        print(user)

8. 关系操作：一对多和多对一

8.1 一对多关系示例

from sqlalchemy import create_engine, String, Integer, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]
    addresses: Mapped[list["Address"]] = relationship(
        "Address", back_populates="user", cascade="all, delete-orphan"
    )

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

class Address(Base):
    __tablename__ = "addresses"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    email: Mapped[str] = mapped_column(String(100))
    user: Mapped[User] = relationship("User", back_populates="addresses")

    def __repr__(self):
        return f"<Address(id={self.id}, email='{self.email}')>"

engine = create_engine("sqlite:///relationship_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 新建用户和多个地址
    user = User(name="赵六", age=35, addresses=[
        Address(email="zhaoliu@example.com"),
        Address(email="zl@company.com")
    ])
    session.add(user)
    session.commit()
    print("创建用户和地址成功")

    # 查询用户及其地址
    user = session.scalars(select(User).where(User.name == "赵六")).first()
    print(f"\n用户：{user.name}")
    print("地址列表：")
    for address in user.addresses:
        print(f"  - {address.email}")

    # 通过地址查用户
    address = session.scalars(select(Address).where(Address.email == "zhaoliu@example.com")).first()
    print(f"\n地址：{address.email}")
    print(f"所属用户：{address.user.name}")

8.2 关系关键说明

ForeignKey ：声明外键
relationship() ：定义模型之间的关系
back_populates ：让关系可双向访问

9. 查询进阶：复杂条件与聚合

9.1 复杂条件查询

from sqlalchemy import create_engine, String, Integer, and_, or_, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]
    city: Mapped[str] = mapped_column(String(50))

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age}, city='{self.city}')>"

engine = create_engine("sqlite:///query_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 添加数据
    users = [
        User(name="张三", age=25, city="北京"),
        User(name="李四", age=30, city="上海"),
        User(name="王五", age=28, city="北京"),
        User(name="赵六", age=35, city="广州"),
    ]
    session.add_all(users)
    session.commit()

    # AND条件
    result1 = session.scalars(select(User).where(and_(User.age > 25, User.city == "北京"))).all()
    print("年龄大于25且城市为北京的用户：")
    for user in result1:
        print(user)

    # OR条件
    result2 = session.scalars(select(User).where(or_(User.age < 25, User.city == "上海"))).all()
    print("\n年龄小于25或城市为上海的用户：")
    for user in result2:
        print(user)

    # LIKE
    result3 = session.scalars(select(User).where(User.name.like("张%"))).all()
    print("\n姓名以'张'开头的用户：")
    for user in result3:
        print(user)

    # IN
    result4 = session.scalars(select(User).where(User.city.in_(["北京", "上海"]))).all()
    print("\n城市为北京或上海的用户：")
    for user in result4:
        print(user)

    # 范围
    result5 = session.scalars(select(User).where(User.age.between(28, 32))).all()
    print("\n年龄在28到32之间的用户：")
    for user in result5:
        print(user)

9.2 聚合查询

from sqlalchemy import create_engine, String, Integer, select, func
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]
    city: Mapped[str] = mapped_column(String(50))

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///query_example.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    # 平均
    avg_age = session.execute(select(func.avg(User.age))).scalar_one()
    print(f"平均年龄：{avg_age:.2f}")

    # 最大
    max_age = session.execute(select(func.max(User.age))).scalar_one()
    print(f"最大年龄：{max_age}")

    # 最小
    min_age = session.execute(select(func.min(User.age))).scalar_one()
    print(f"最小年龄：{min_age}")

    # 总和
    total_age = session.execute(select(func.sum(User.age))).scalar_one()
    print(f"年龄总和：{total_age}")

    # 计数
    user_count = session.execute(select(func.count(User.id))).scalar_one()
    print(f"用户总数：{user_count}")

    # 分组统计
    city_stats = session.execute(
        select(
            User.city,
            func.count(User.id).label("count"),
            func.avg(User.age).label("avg_age")
        ).group_by(User.city)
    ).all()
    print("\n按城市分组统计：")
    for city, count, avg_age in city_stats:
        print(f"  {city}: {count}人, 平均年龄{avg_age:.2f}")

10. 事务管理

事务：一组要么全成功、要么全失败的数据库操作。推荐直接用Session上下文自动管理。

10.1 基本事务操作（推荐）

from sqlalchemy import create_engine, String, Integer
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int]

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', age={self.age})>"

engine = create_engine("sqlite:///transaction_example.db", echo=True)
Base.metadata.create_all(engine)

try:
    with Session(engine) as session:
        user1 = User(name="用户1", age=20)
        user2 = User(name="用户2", age=25)
        session.add_all([user1, user2])
        session.commit()
        print("事务提交成功")
except Exception as e:
    print(f"发生错误：{e}")

10.2 使用上下文管理器（最佳实践）

SQLAlchemy 2.0+ 推荐直接使用 with Session(engine) as session，无须自制器。

with Session(engine) as session:
    user = User(name="测试用户", age=30)
    session.add(user)
    session.commit()
print("操作完成")

11. 常见错误与最佳实践

11.1 常见错误

错误 1：忘记提交事务

# 错误示例
with Session(engine) as session:
    user = User(name="张三", age=25)
    session.add(user)
    # 没有 commit()，数据不会写入数据库！

# 正确示例
with Session(engine) as session:
    user = User(name="张三", age=25)
    session.add(user)
    session.commit()

错误 2：忘记关闭会话

2.0+ 推荐始终用上下文，不手动close：

# 错误示例（手动close易忘）
session = Session(engine)
user = User(name="张三", age=25)
session.add(user)
session.commit()
# 忘记 close()

# 推荐写法（自动管理）
with Session(engine) as session:
    user = User(name="张三", age=25)
    session.add(user)
    session.commit()

错误 3：循环中频繁提交

# 错误示例：极低效
with Session(engine) as session:
    for i in range(100):
        user = User(name=f"用户{i}", age=20+i)
        session.add(user)
        session.commit()  # 每次循环都Commit

# 推荐：批量提交
with Session(engine) as session:
    users = [User(name=f"用户{i}", age=20 + i) for i in range(100)]
    session.add_all(users)
    session.commit()

11.2 最佳实践

始终用Session上下文管理器 ，自动关闭
批量操作 推荐 add_all()，避免循环中多次提交
异常处理 用try-except确保事务安全
模型定义分离 ，生产代码将模型单独放
类型提示 让代码更可维护

12. 总结

12.1 核心概念回顾

Engine ：数据库引擎
Base ：所有模型基类
Session ：操作入口
Model ：数据表 Python 类
Relationship ：表间关系

12.2 基本操作流程

创建 Engine
定义 Base
定义 Model（继承 Base）
创建表
用 with Session(engine) as session:
执行 CRUD，事务提交
结束自动关闭

1. SQLAlchemy 是什么？​

1.1 为什么需要 SQLAlchemy？​

1.2 SQLAlchemy 的核心优势​

2. 前置知识​

2.1 什么是 ORM？​

2.2 数据库基础术语​

2.3 Python 基础​

3. 安装与环境准备​

3.1 安装 SQLAlchemy​

3.2 选择数据库​

4. 核心概念：Engine、Base、Session​

4.1 Engine（引擎）​

4.2 Base（声明式基类）​

4.3 Session（会话）​

5. 第一个完整示例：从零开始​

6. 定义数据模型​

6.1 基本模型定义​

6.2 常用列类型​

6.3 列参数​

7. 基本 CRUD 操作​

7.1 创建（Create）：添加数据​

7.2 读取（Read）：查询数据​

7.3 更新（Update）：修改数据​

7.4 删除（Delete）：删除数据​

8. 关系操作：一对多和多对一​

8.1 一对多关系示例​

8.2 关系关键说明​

9. 查询进阶：复杂条件与聚合​

9.1 复杂条件查询​

9.2 聚合查询​

10. 事务管理​

10.1 基本事务操作（推荐）​

10.2 使用上下文管理器（最佳实践）​

11. 常见错误与最佳实践​

11.1 常见错误​

错误 1：忘记提交事务​

错误 2：忘记关闭会话​

错误 3：循环中频繁提交​

11.2 最佳实践​

12. 总结​

12.1 核心概念回顾​

12.2 基本操作流程​