惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

The GitHub Blog
The GitHub Blog
K
Kaspersky official blog
Stack Overflow Blog
Stack Overflow Blog
Blog — PlanetScale
Blog — PlanetScale
Recorded Future
Recorded Future
Engineering at Meta
Engineering at Meta
U
Unit 42
D
Docker
I
InfoQ
D
DataBreaches.Net
Google DeepMind News
Google DeepMind News
N
Netflix TechBlog - Medium
C
Check Point Blog
The Cloudflare Blog
美团技术团队
V
Vulnerabilities – Threatpost
博客园_首页
T
Threat Research - Cisco Blogs
Google DeepMind News
Google DeepMind News
Attack and Defense Labs
Attack and Defense Labs
A
Arctic Wolf
IT之家
IT之家
让小产品的独立变现更简单 - ezindie.com
让小产品的独立变现更简单 - ezindie.com
Exploit-DB.com RSS Feed
Exploit-DB.com RSS Feed
T
Troy Hunt's Blog
cs.CL updates on arXiv.org
cs.CL updates on arXiv.org
人人都是产品经理
人人都是产品经理
C
Cyber Attacks, Cyber Crime and Cyber Security
钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知
Security Archives - TechRepublic
Security Archives - TechRepublic
S
Schneier on Security
Apple Machine Learning Research
Apple Machine Learning Research
MyScale Blog
MyScale Blog
P
Privacy International News Feed
云风的 BLOG
云风的 BLOG
Recent Commits to openclaw:main
Recent Commits to openclaw:main
T
The Blog of Author Tim Ferriss
GbyAI
GbyAI
The Last Watchdog
The Last Watchdog
D
Darknet – Hacking Tools, Hacker News & Cyber Security
aimingoo的专栏
aimingoo的专栏
P
Proofpoint News Feed
The Register - Security
The Register - Security
博客园 - 三生石上(FineUI控件)
Forbes - Security
Forbes - Security
NISL@THU
NISL@THU
Y
Y Combinator Blog
T
Threatpost
Microsoft Azure Blog
Microsoft Azure Blog
L
Lohrmann on Cybersecurity

又见苍岚

COLMAP PatchMatch Stereo 算法详解 事件驱动的状态机框架:从理论到工程实践 Git 在国内网络环境下无法 Push 的排查与修复 —— 配置 Clash 代理 分段五次多项式插值原理详解 路径插值方法深度对比研究 Claude Code 使用指南 OpenClaw 记忆管理与技能创建指南 CBS(Conflict-Based Search)算法详解 A* 算法及其变种详解 OpenClaw 配置多 Agents Windows Powershell 无法加载文件,因为在此系统上禁止运行脚本问题的解决方案 MaxClaw 安装流程 大模型 AI 名词介绍 AList 网盘聚合工具简介 Protobuf 简介与测试 Claude Code 简介以及 GLM 4.7 模型接入 Github 歌词下载工具 163MusicLyrics Python __getattr__ 懒加载 Python TypedDict 机器人仿真平台 Gazebo 安装记录 机器人仿真平台 Gazebo 简介 多机器人路径规划问题(Multi-Agent Path Finding, MAPF)简介 Python exifread 读取修改过的 jpeg 信息错误问题修复 3D 坐标系变换的理解 3D 旋转矩阵基本概念 MongoDB Compass 介绍 Python 环境管理工具 uv Flutter 开发指南 Snipaste 安装下载与黑屏问题解决方案 全局路径规划算法记录 2025 Python 版本性能测试 Flutter Hello World Flutter 安装环境配置 Ubuntu VMware 硬盘扩容后 SMBus Host controller not enabled 报错问题解决 Python NetworkX 教程 Docker GPU 报错 - Failed to initialize NVML Unknown Error 解决方案 Python matplotlib 图表绘制 cuda-toolkit 安装替代 Cuda 与 Cudnn Jinja2 Python 利用 docxtpl 和 Jinja2 生成基于模板的 Word 文档 Docker 实现 CPU 核心隔离 LoFTR 基于 Transformer 的特征提取匹配算法 OmniGlue 特征匹配 SuperGlue 使用图神经网络学习特征匹配 Ubuntu 下将 xlsx 文件按照 sheet 转换为 图片 Python FastAPI 教程 openwrt 软路由配置安装 Nav2 地图文件(PGM/YAML)规范标准 3D OBJ 模型转换为 glb 瓦片格式 Python 源码 Redis 数据库介绍 Ubuntu 22.04 内核自动升级导致 MongoDB 7.0.12 错误记录 ubuntu 20.04 安装 ROS Noetic ubuntu 18.04 安装 ROS Melodic VMware Workstation Pro 个人免费版下载、安装、使用指南 Hybrid A-star 路径规划 Reeds-Shepp 曲线 Dubins 曲线 Linux kvm 虚拟机网络不通的问题解决方法 Ubuntu 自动内存清理 BiliBili 缓存视频转 mp4 Python 求解线性规划 3D Gaussian Splatting 官方源码实践记录 ImageMagick 教程 Ubuntu 22.04 安装 Colmap 对数几率 odds Ubuntu nmcli 网络管理工具使用指南 SuperPoint 自监督深度学习特征点提取 SyncTV Music Tag Web 在线音乐信息整理工具 ncm 格式转 mp3 MusicBrainz 音乐元数据百科数据库 Ubuntu 网络流量监控工具 私人云音乐平台 Navidrome 入门 手眼标定 四元数(Quaternions) OHTTPS 实现免费自动 https 证书申请、更新、部署 ubuntu 22.04 安装 CloudCompare 单机 KVM 虚拟机冷迁移 Ubuntu 22.04 使用 mdadm 实现软 raid 小鱼 一键安装 ROS-humble Fluid -46- 基于 Simpletex API 构建公式识别页面 公式识别 API 简介 -- Simpletex 使用 Python web 部署库 waitress 3D Gaussian Splatting for Real-Time Radiance Field Rendering Ubuntu Swap 简介与空间扩展 Ubuntu 24.04 安装 forticlient Clash Verge 使用 MongoDB 7.0.17 集群 Docker 构建源码 Error code - 2013. Lost connection to MySQL server during query 问题解决 Python 日志记录库 loguru 使用指北 Python 实现 Web 日志查看服务 MySQL LOAD DATA LOCAL INFILE 极速数据加载 Image size exceeds limit of 89478485 pixels 解决方案 Docker 使用 NVIDIA GPU 驱动错误解决 阿里云 docker 镜像仓库 Ubuntu中没有wired connected的解决方案 MinIO 简介 subconverter 代理订阅格式转换 修复 node –openssl-legacy-provider is not allowed in NODE_OPTIONS 错误 Ubuntu 使用 Clash For Linux 客户端
Python 使用 SQLAlchemy
Yiwei Zhang · 2025-07-24 · via 又见苍岚

本文介绍了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,表示一行记录,比如,包含idnameuser表:

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

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

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

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/