1. SQLAlchemy 是什么？

SQLAlchemy 是 Python 最流行的数据库 ORM 框架之一。

它提供两种使用方式：

Core
- 更接近原生 SQL
- 手写 SQL 表达式
ORM（Object Relational Mapping）
- 用 Python 类表示数据库表
- 用对象操作数据

日常 Web 开发（FastAPI 等）通常使用 ORM。

核心关系：

数据库
  |
 Engine
  |
 Session
  |
 ORM对象(Model)

2. 安装

1	pip install sqlalchemy

SQLite 不需要额外安装驱动。

3. SQLAlchemy 三个核心概念

3.1 Base：ORM 模型基类

Base 是所有数据表模型的父类。

例如：

数据库表：

users

id
name
age
email

Python 中：

1 2	class User(Base): ...

继承 Base 后，SQLAlchemy 才知道：

这个类需要映射成数据库表。

创建 Base：

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

以后所有模型：

1 2	class User(Base): __tablename__ = "users"

都继承它。

3.2 Engine：数据库连接入口

Engine 负责：

连接数据库
管理数据库连接池
执行 SQL

创建：

from sqlalchemy import create_engine


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

解释：

1	sqlite:///database.db

表示：

当前目录创建：

1	database.db

echo：

echo=True

表示打印 SQL：

例如：

1	SELECT * FROM users;

方便调试。

3.3 Session：数据库操作会话

Session 是 ORM 操作数据库的核心。

它负责：

查询
新增
修改
删除
事务管理

例如：

from sqlalchemy.orm import Session


with Session(engine) as session:

    ...

流程：

创建Session

      ↓

操作对象

      ↓

commit提交

      ↓

数据库保存

Session 类似一个数据库工作单元。

4. 创建数据表

定义模型

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


class User(Base):

    __tablename__ = "users"


    id: Mapped[int] = mapped_column(
        primary_key=True
    )


    name: Mapped[str] = mapped_column(
        String(50)
    )


    age: Mapped[int] = mapped_column(
        Integer
    )


    email: Mapped[str] = mapped_column(
        String(100)
    )

解释：

tablename

对应数据库表名：

1	__tablename__ = "users"

数据库：

users表

mapped_column

表示字段。

例如：

1
2
3

id: Mapped[int] = mapped_column(
    primary_key=True
)

对应：

1	id INTEGER PRIMARY KEY

5. 创建表

1	Base.metadata.create_all(engine)

作用：

读取所有继承 Base 的模型：

1
2
3

User
Order
Product

然后创建数据库表。

例如：

1
2
3

database.db

users表

6. Session 工厂（项目常用）

实际项目一般不会：

1	Session(engine)

而是：

from sqlalchemy.orm import sessionmaker

# 这里没有创建连接，只是个机器
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False
)

以后：

1 2	with SessionLocal() as session: # 这一步，机器才吐出一个全新的 session ...

为什么？

因为：

一个请求

    |
    创建一个Session

    |
    操作数据库

    |
    关闭Session

7. CRUD 操作

C - Create 创建

新增数据：

with Session(engine) as session:

    user = User(
        name="Tom",
        age=18,
        email="tom@qq.com"
    )


    session.add(user)


    session.commit()

流程：

创建对象

↓

add加入Session

↓

commit保存

批量新增：

users = [
    User(
        name="Alice",
        age=20,
        email="a@qq.com"
    ),

    User(
        name="Bob",
        age=22,
        email="b@qq.com"
    )
]


session.add_all(users)

session.commit()

R - Read 查询

SQLAlchemy 2.x 推荐：

select()

导入：

1	from sqlalchemy import select

查询全部

with Session(engine) as session:


    stmt = select(User)


    users = session.scalars(
        stmt
    ).all()


    print(users)

对应 SQL：

1 2	SELECT * FROM users;

根据 id 查询

user = session.get(
    User,
    1
)

对应：

1
2
3

SELECT *
FROM users
WHERE id=1;

条件查询

年龄大于18：

stmt = select(User).where(
    User.age > 18
)


users = session.scalars(
    stmt
).all()

SQL：

1
2
3

SELECT *
FROM users
WHERE age > 18;

多条件查询

stmt = select(User).where(
    User.age > 18,
    User.name == "Tom"
)

等价：

1 2	WHERE age>18 AND name='Tom'

U - Update 修改

ORM方式修改

先查询对象：

user = session.get(
    User,
    1
)


user.age = 25


session.commit()

SQLAlchemy 自动生成：

1
2
3

UPDATE users
SET age=25
WHERE id=1;

批量更新

from sqlalchemy import update


stmt = (
    update(User)
    .where(
        User.name=="Tom"
    )
    .values(
        age=30
    )
)


session.execute(stmt)

session.commit()

D - Delete 删除

删除对象

user = session.get(
    User,
    1
)


session.delete(user)


session.commit()