本文最后更新于：2025年7月30日 下午

本文介绍了Python的ORM工具SQLAlchemy的使用方法，包括如何建立与数据库连接、建立映射关系、单表及多表的CRUD操作。通过示例展示了如何使用query()方法进行灵活的查询，以及如何处理多表之间的关系。此外，还演示了如何创建、修改和删除记录。。

简介

SQLAlchemy 是 Python 著名的 ORM(Object Relational Mapping) 工具包。通过 ORM，开发者可以用面向对象的方式来操作数据库，不再需要编写 SQL 语句。本篇不解释为什么要使用 ORM，主要讲解 SQLAlchemy 的用法。SQLAlchemy 支持多种数据库，除 sqlite 外，其它数据库需要安装第三方驱动。本篇以 sqlite 数据库为例进行说明。

关于ORM：

全称 Object Relational Mapping（对象关系映射）。

特点是操纵Python对象而不是SQL查询，也就是在代码层面考虑的是对象，而不是SQL，体现的是一种程序化思维，这样使得Python程序更加简洁易读。

具体的实现方式是将数据库表转换为Python类，其中数据列作为属性，数据库操作作为方法。

优点：

  1. 简洁易读：将数据表抽象为对象（数据模型），更直观易读
  2. 可移植：封装了多种数据库引擎，面对多个数据库，操作基本一致，代码易维护
  3. 更安全：有效避免SQL注入

SQLAlchemy分成三部分：

  1. ORM，就是我们用类来表示数据库schema的那部分
  2. SQLAlchemy Core，就是一些基础的操作，例如 update, insert 等等，也可以直接使用这部分来进行操作，但是它们写起来没有ORM那么自然
  3. DBAPI，这部分就是数据库驱动

它们的关系如下

安装 SQLAlchemy

使用pip工具安装SQLAlchemy：

1
pip install sqlalchemy

ORM

数据库表是一个二维表，包含多行多列。把一个表的内容用Python的数据结构表示出来的话，可以用一个list表示多行，list的每一个元素是tuple，表示一行记录，比如，包含id和name的user表：

1
2
3
4
5
[
    ('1', 'Michael'),
    ('2', 'Bob'),
    ('3', 'Adam')
]

Python的DB-API返回的数据结构就是像上面这样表示的。

但是用tuple表示一行很难看出表的结构。如果把一个tuple用class实例来表示，就可以更容易地看出表的结构来：

1
2
3
4
5
6
7
8
9
10
class User(object):
    def __init__(self, id, name):
        self.id = id
        self.name = name

[
    User('1', 'Michael'),
    User('2', 'Bob'),
    User('3', 'Adam')
]

这就是传说中的ORM技术：Object-Relational Mapping，把关系数据库的表结构映射到对象上。

简单示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import contextlib
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import (
    create_engine,
    Column,
    Integer,
    DateTime,
    String,
)
from config import config  # config模块里有自己写的配置，我们可以换成别的，注意下面用到config的地方也要一起换

engine = create_engine(
    config.SQLALCHEMY_DATABASE_URI,  # SQLAlchemy 数据库连接串，格式见下面
    echo=bool(config.SQLALCHEMY_ECHO),  # 是不是要把所执行的SQL打印出来，一般用于调试
    pool_size=int(config.SQLALCHEMY_POOL_SIZE),  # 连接池大小
    max_overflow=int(config.SQLALCHEMY_POOL_MAX_SIZE),  # 连接池最大的大小
    pool_recycle=int(config.SQLALCHEMY_POOL_RECYCLE),  # 多久时间主动回收连接，见下注释
)
Session = sessionmaker(bind=engine)
Base = declarative_base(engine)


class BaseMixin:
    """model的基类,所有model都必须继承"""
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at = Column(DateTime, nullable=False, default=datetime.datetime.now, onupdate=datetime.datetime.now, index=True)
    deleted_at = Column(DateTime)  # 可以为空, 如果非空, 则为软删


@contextlib.contextmanager
def get_session():
    s = Session()
    try:
        yield s
        s.commit()
    except Exception as e:
        s.rollback()
        raise e
    finally:
        s.close()


class User(Base, BaseMixin):
    __tablename__ = "user"

    Name = Column(String(36), nullable=False)
    Phone = Column(String(36), nullable=False, unique=True)

我们注意上面的几点：

  • pool_recycle，设置主动回收连接的时长，如果不设置，那么可能会遇到数据库主动断开连接的问题，例如MySQL通常会为连接设置 最大生命周期为八小时，如果没有通信，那么就会断开连接。因此不设置此选项可能就会遇到 MySQL has gone away 的报错。

  • engine，engine是SQLAlchemy 中位于数据库驱动之上的一个抽象概念，它适配了各种数据库驱动，提供了连接池等功能。其用法就是 如上面例子中，

    engine = create_engine(<数据库连接串>)，数据库连接串的格式是 dialect+driver://username:password@host:port/database?参数这样的，dialect 可以是mysql, postgresql, oracle, mssql, sqlite，后面的 driver 是驱动，比如MySQL的驱动pymysql， 如果不填写，就使用默认驱动。再往后就是用户名、密码、地址、端口、数据库、连接参数了，我们来看几个例子：

    • MySQL: engine = create_engine('mysql+pymysql://scott:tiger@localhost/foo?charset=utf8mb4')
    • PostgreSQL: engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')
    • Oracle: engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
    • MS SQL: engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')
    • SQLite: engine = create_engine('sqlite:////absolute/path/to/foo.db')
    • 详见：https://docs.sqlalchemy.org/en/13/core/engines.html

  • Session，Session的意思就是会话，也就是说，是一个逻辑组织的概念，因此，这需要靠你的业务逻辑来划分哪些操作使用同一个Session， 哪些操作又划分为不同的业务操作，详见 这里。 举个简单的例子，以web应用为例，一个请求里共用一个Session就是一个好的例子，一个异步任务执行过程中使用一个Session也是一个例子。 但是注意，不能直接使用Session，而是使用Session的实例，借助上面的代码，我们可以直接这样写：

    1
    2
    with get_session() as s:
    print(s.query(User).first())

  • Base，Base是ORM中的一个基类，通过集成Base，我们才能方便的使用一些基本的查询，例如 s.query(User).filter_by(User.name="nick").first()。

  • BaseMixin，BaseMixin是我自己定义的一些通用的表结构，通过Mixin的方式集成到类里，比如上面的定义，我们常见的表结构里，都会有 ID、创建时间，更新时间，软删除标志等等，我们把它作为一个独立的类，这样通过继承即可获得相关表属性，省得重复写多次。

表的设计

表的设计通常就如 User 表一样：

1
2
3
4
5
class User(Base, BaseMixin):
    __tablename__ = "user"

    Name = Column(String(36), nullable=False)
    Phone = Column(String(36), nullable=False, unique=True)

首先使用 __tablename__ 自定义表名，接着写各个表中的属性，也就是对应在数据库表中的列(column)，常见的类型有：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
$ egrep '^class ' ~/.pyenv/versions/3.6.0/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py
class _LookupExpressionAdapter(object):
class Concatenable(object):
class Indexable(object):
class String(Concatenable, TypeEngine):
class Text(String):
class Unicode(String):
class UnicodeText(Text):
class Integer(_LookupExpressionAdapter, TypeEngine):
class SmallInteger(Integer):
class BigInteger(Integer):
class Numeric(_LookupExpressionAdapter, TypeEngine):
class Float(Numeric):
class DateTime(_LookupExpressionAdapter, TypeEngine):
class Date(_LookupExpressionAdapter, TypeEngine):
class Time(_LookupExpressionAdapter, TypeEngine):
class _Binary(TypeEngine):
class LargeBinary(_Binary):
class Binary(LargeBinary):
class SchemaType(SchemaEventTarget):
class Enum(Emulated, String, SchemaType):
class PickleType(TypeDecorator):
class Boolean(Emulated, TypeEngine, SchemaType):
class _AbstractInterval(_LookupExpressionAdapter, TypeEngine):
class Interval(Emulated, _AbstractInterval, TypeDecorator):
class JSON(Indexable, TypeEngine):
class ARRAY(SchemaEventTarget, Indexable, Concatenable, TypeEngine):
class REAL(Float):
class FLOAT(Float):
class NUMERIC(Numeric):
class DECIMAL(Numeric):
class INTEGER(Integer):
class SMALLINT(SmallInteger):
class BIGINT(BigInteger):
class TIMESTAMP(DateTime):
class DATETIME(DateTime):
class DATE(Date):
class TIME(Time):
class TEXT(Text):
class CLOB(Text):
class VARCHAR(String):
class NVARCHAR(Unicode):
class CHAR(String):
class NCHAR(Unicode):
class BLOB(LargeBinary):
class BINARY(_Binary):
class VARBINARY(_Binary):
class BOOLEAN(Boolean):
class NullType(TypeEngine):
class MatchType(Boolean):

常见操作

我们来看看使用SQLAlchemy完成常见的操作，例如增删查改：

常见查询操作

  • SELECT * FROM user 应该这样写：

    1
    2
    with get_session() as s:
    print(s.query(User).all())

  • SELECT * FROM user WHERE name='nick' 应该这样写：

    1
    2
    3
    with get_session() as s:
    print(s.query(User).filter_by(User.name='nick').all())
    print(s.query(User).filter(User.name == 'nick').all())  # 这样写是等同效果的

  • SELECT * FROM user WHERE name='nick' LIMIT 1 应该这样写：

    1
    2
    with get_session() as s:
    print(s.query(User).filter_by(User.name='nick').first())

如果需要加判定，例如确保只有一条数据，那就把 first() 替换为 one()，如果确保一行或者没有，那就写 one_or_none()。

  • SELECT * FROM user ORDER BY id DESC LIMIT 1 应该这样写：

    1
    2
    with get_session() as s:
    print(s.query(User).order_by(User.id.desc()).first())

  • SELECT * FROM user ORDER BY id DESC LIMIT 1 OFFSET 20 应该这样写：

    1
    2
    with get_session() as s:
    print(s.query(User).order_by(User.id.desc()).offset(20).first())

常见删除操作

  • DELETE FROM user 应该这样写：

    1
    2
    with get_session() as s:
    s.query(User).delete()

  • DELETE FROM user WHERE name='nick'：

    1
    2
    with get_session() as s:
    s.query(User).filter_by(User.name='nick').delete()

  • DELETE FROM user WHERE name='nick' LIMIT 1：

    1
    2
    with get_session() as s:
    s.query(User).filter_by(User.name='nick').limit(1).delete()

更新语句

  • UPDATE user SET name='nick'：

    1
    2
    with get_session() as s:
    s.query(User).update({'name': 'nick'})

  • UPDATE user SET name='nick' WHERE id=1：

    1
    2
    with get_session() as s:
    s.query(User).filter_by(User.id=1).update({'name': 'nick'})

也可以通过更改实例的属性，然后提交：

1
2
3
4
with get_session() as s:
    user = s.query(User).filter_by(User.id=1).one()
    user.name = 'nick'
    s.commit()

插入语句

这个就简单了，实例化对象，然后 session.add，最后提交：

1
2
3
4
with get_session() as s:
    user = User()
    s.add(user)
    s.commit()

连表

SQLAlchemy 中可以直接使用join语句：

1
2
with get_session() as s:
    s.query(Customer).join(Invoice).filter(Invoice.amount == 8500)

可以是这么几种写法：

1
2
3
4
query.join(Address, User.id==Address.user_id)    # explicit condition
query.join(User.addresses)                       # specify relationship from left to right
query.join(Address, User.addresses)              # same, with explicit target
query.join('addresses')                          # same, using a string

数据库migration

我们使用alembic来做数据库migration，首先安装：

1
2
$ pip install alembic
$ alembic init alembic  # 此处 alembic init 后接的是保存migration的文件夹名称

然后我们要修改 alembic/env.py (假设你设置的保存migration的文件夹名称就是 alembic)，将对应部分修改成如下：

1
2
3
4
5
config.set_main_option(
    'sqlalchemy.url', config.SQLALCHEMY_DATABASE_URI
)
target_metadata = Base.metadata  # 从任意一个我们的model可以拿到总的Base
engine = target_metadata.bind

因为SQLAlchemy会把表的信息存储在 metadata 里，而我们都继承了 Base，因此可以 通过 Base.metadata 来拿到所有表的信息，这样子alembic才能够拿到表的结构，然后和 数据库进行对比，生成migration脚本：

1
$ alembic revision --autogenerate -m '本次migration的信息，相当于git提交时的评论'

应用举例

尝试在 MySQL 的 test 数据库中创建的user表，用SQLAlchemy来试试：

第一步，导入SQLAlchemy，并初始化DBSession：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 导入:
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import declarative_base

# 创建对象的基类:
Base = declarative_base()

# 定义User对象:
class User(Base):
    # 表的名字:
    __tablename__ = 'user'

    # 表的结构:
    id = Column(String(20), primary_key=True)
    name = Column(String(20))

# 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://root:password@localhost:3306/test')
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)

以上代码完成SQLAlchemy的初始化和具体每个表的class定义。如果有多个表，就继续定义其他class，例如School：

1
2
3
4
class School(Base):
    __tablename__ = 'school'
    id = ...
    name = ...

create_engine()用来初始化数据库连接。SQLAlchemy用一个字符串表示连接信息：

1
'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'

你只需要根据需要替换掉用户名、口令等信息即可。

下面，我们看看如何向数据库表中添加一行记录。

由于有了ORM，我们向数据库表中添加一行记录，可以视为添加一个User对象：

1
2
3
4
5
6
7
8
9
10
# 创建session对象:
session = DBSession()
# 创建新User对象:
new_user = User(id='5', name='Bob')
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
session.commit()
# 关闭session:
session.close()

可见，关键是获取session，然后把对象添加到session，最后提交并关闭。DBSession对象可视为当前数据库连接。

如何从数据库表中查询数据呢？有了ORM，查询出来的可以不再是tuple，而是User对象。SQLAlchemy提供的查询接口如下：

1
2
3
4
5
6
7
8
9
# 创建Session:
session = DBSession()
# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()
# 打印类型和对象的name属性:
print('type:', type(user))
print('name:', user.name)
# 关闭Session:
session.close()

运行结果如下：

1
2
type: <class '__main__.User'>
name: Bob

可见，ORM就是把数据库表的行与相应的对象建立关联，互相转换。

由于关系数据库的多个表还可以用外键实现一对多、多对多等关联，相应地，ORM框架也可以提供两个对象之间的一对多、多对多等功能。

例如，如果一个User拥有多个Book，就可以定义一对多关系如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class User(Base):
    __tablename__ = 'user'

    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # 一对多:
    books = relationship('Book')

class Book(Base):
    __tablename__ = 'book'

    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # “多”的一方的book表是通过外键关联到user表的:
    user_id = Column(String(20), ForeignKey('user.id'))

当我们查询一个User对象时，该对象的books属性将返回一个包含若干个Book对象的list。

参考资料



文章链接：
https://www.zywvvd.com/notes/coding/python/python-sqlalchemy/python-sqlalchemy/

“觉得不错的话，给点打赏吧 ୧(๑•̀⌄•́๑)૭”

微信二维码

微信支付

支付宝二维码

支付宝支付

Coding > Python
#Python #Coding
Python 使用 SQLAlchemy
https://www.zywvvd.com/notes/coding/python/python-sqlalchemy/python-sqlalchemy/
作者
Yiwei Zhang
发布于
2025年7月24日
许可协议