SQLAlchemy 2.0 查询使用指南

SQLAlchemy 2.0 查询使用指南

1. 环境设置

首先，需要安装 SQLAlchemy 2.0 版本。假设你使用的是 SQLite 数据库，可以通过以下命令安装 SQLAlchemy：

pip install sqlalchemy

接着，我们创建数据库连接并初始化会话：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker# 创建数据库引擎
engine = create_engine('sqlite:///example.db', echo=True)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

2. 定义数据模型

SQLAlchemy 使用 ORM (对象关系映射) 使得 Python 对象和数据库表之间实现映射。以下是定义 User​ 和 Address​ 表的模型示例：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationshipBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String)age = Column(Integer)# 一对多关系，用户可以有多个地址addresses = relationship('Address', back_populates='user', lazy='select')class Address(Base):__tablename__ = 'addresses'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))email = Column(String)# 反向关系user = relationship('User', back_populates='addresses')

在这个模型中，User​ 表和 Address​ 表通过 user_id​ 建立了外键关联，用户可以有多个地址。

3. 基本查询操作

SQLAlchemy 2.0 提供了一个更简洁和一致的查询接口。以下是一些常见的查询操作。

3.1 查询所有记录

from sqlalchemy import selectstmt = select(User)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users;

3.2 查询特定字段

查询 name​ 和 age​ 字段：

stmt = select(User.name, User.age)
result = session.execute(stmt).all()for name, age in result:print(name, age)

SQL 原生查询：

SELECT name, age FROM users;

3.3 查询带条件的记录

查询年龄大于 30 的用户：

stmt = select(User).where(User.age > 30)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users WHERE age > 30;

3.4 排序查询

根据年龄降序排列查询用户：

stmt = select(User).order_by(User.age.desc())
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users ORDER BY age DESC;

3.5 限制查询结果

限制返回的记录数为 5：

stmt = select(User).limit(5)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users LIMIT 5;

4. 关联查询（Join）

4.1 懒查询 (Lazy Loading)

懒查询是 SQLAlchemy 默认的加载方式，只有在访问关联属性时才会执行查询。它通过延迟加载来避免不必要的查询，但是可能导致 “N+1 查询问题”。

# 查询用户，并访问用户的地址
user = session.query(User).first()  # 执行查询，不查询 addresses
print(user.name)  # 输出用户名称addresses = user.addresses  # 当访问 addresses 时，SQLAlchemy 会查询 addresses 表
for address in addresses:print(address.email)

SQL 原生查询：

SELECT * FROM users WHERE id = 1;
-- 查询用户的地址
SELECT * FROM addresses WHERE user_id = 1;

4.2 预加载（Eager Loading）

预加载是通过 JOIN 或子查询一次性加载所有相关数据。常见的预加载方法有 joinedload​、subqueryload​ 和 selectinload​。

4.2.1 joinedload​

​joinedload​ 会通过 JOIN 一次性加载所有相关数据，适用于关联数据量少的情况。

from sqlalchemy.orm import joinedloadstmt = select(User).options(joinedload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查询：

SELECT * FROM users
JOIN addresses ON users.id = addresses.user_id;
4.2.2 subqueryload​

​subqueryload​ 使用子查询来加载关联数据，适用于关联数据较多的情况。

from sqlalchemy.orm import subqueryloadstmt = select(User).options(subqueryload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查询：

SELECT * FROM users;
-- 查询 addresses 表的所有数据
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);
4.2.3 selectinload​

​selectinload​ 适用于多对一和一对多的关系，能够通过一次查询批量加载所有关联数据。

from sqlalchemy.orm import selectinloadstmt = select(User).options(selectinload(User.addresses))
result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

SQL 原生查询：

SELECT * FROM users;
-- 查询 addresses 表
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.3 延迟加载与预加载的比较

特性懒查询 (Lazy Loading)预加载 (Eager Loading)
数据加载方式延迟加载，直到访问关联属性时才查询一次性加载所有关联数据
查询次数可能多次查询（N+1 查询问题）一次性查询，通常只发出少量查询
性能对于小数据量高效对于复杂查询避免 N+1 问题，适用于大量关联数据
常用方法无​joinedload​、subqueryload​、selectinload​

5. 高级查询

5.1 聚合查询

SQLAlchemy 支持聚合函数，如 count​、sum​、avg​ 等。以下是计算用户数量的例子：

from sqlalchemy import funcstmt = select(func.count(User.id))
result = session.execute(stmt).scalar()
print(f"User Count: {result}")

SQL 原生查询：

SELECT COUNT(id) FROM users;

5.2 分组查询

通过 group_by​ 方法进行分组查询：

stmt = select(User.age, func.count(User.id)).group_by(User.age)
result = session.execute(stmt).all()for age, count in result:print(f"Age: {age}, Count: {count}")

SQL 原生查询：

SELECT age, COUNT(id) FROM users GROUP BY age;

5.3 子查询

通过子查询来执行更复杂的查询：

subquery = select(func.max(User.age)).scalar_subquery()
stmt = select(User).where(User.age == subquery)
result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users WHERE age = (SELECT MAX(age) FROM users);

6. 事务管理

SQLAlchemy 2.0 自动管理事务，但你也可以显式地管理事务。例如：

from sqlalchemy.exc import SQLAlchemyErrortry:new_user = User(name="David", age=28)session.add(new_user)session.commit()  # 提交事务
except SQLAlchemyError:session.rollback()  # 回滚事务print("事务失败，已回滚")

7. 总结

SQLAlchemy 2.0 提供了强大的 ORM 功能和灵活的查询接口。合理选择懒查询和预加载策略可以有效避免性能问题，特别是对于关系复杂的数据模型，预加载能帮助避免 N+1 查询问题。通过使用 select()​、joinedload​、subqueryload​ 等方法，我们可以优化查询性能，提高数据操作的效率。

当然，以下是更详细的 SQLAlchemy 2.0 查询使用指南，包括懒查询（Lazy Loading）、预加载（Eager Loading）以及异步 session​ 使用示例。我们将通过详细的案例来解释这些概念，帮助团队成员全面理解 SQLAlchemy 2.0 的使用方式。

SQLAlchemy 2.0 查询使用指南

1. 环境设置

首先，确保安装了 SQLAlchemy 2.0：

pip install sqlalchemy

对于异步支持，确保你安装了 asyncpg​（适用于 PostgreSQL）或 aiomysql​（适用于 MySQL）等异步数据库驱动。

pip install sqlalchemy[asyncio] asyncpg

1.1 创建数据库引擎和会话

数据库连接使用 create_engine()​ 来创建数据库引擎，会话使用 sessionmaker()​ 来创建。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker# 创建数据库引擎
engine = create_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)# 创建异步会话
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine# 异步引擎
engine = create_async_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)# 异步会话
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

2. 定义数据模型

SQLAlchemy ORM 通过 declarative_base()​ 定义数据库模型。这里我们定义 User​ 和 Address​ 模型。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationshipBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String)age = Column(Integer)# 一对多关系addresses = relationship('Address', back_populates='user', lazy='select')class Address(Base):__tablename__ = 'addresses'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))email = Column(String)user = relationship('User', back_populates='addresses')

2.1 解释模型

在这个模型中，User​ 表有一个一对多关系与 Address​ 表关联，User​ 表的 addresses​ 属性代表用户的多个地址，而 Address​ 表的 user​ 属性反向关联用户。

3. 基本查询操作

SQLAlchemy 2.0 提供了更加直观的查询接口。以下是一些常见的查询操作。

3.1 查询所有记录

查询 User​ 表中的所有用户数据：

from sqlalchemy import select# 创建查询语句
stmt = select(User)# 执行查询
with session.begin():result = session.execute(stmt).scalars().all()# 输出结果
for user in result:print(user.name, user.age)

原生 SQL 查询：

SELECT * FROM users;

3.2 查询特定字段

查询 name​ 和 age​ 字段的用户数据：

stmt = select(User.name, User.age)with session.begin():result = session.execute(stmt).all()for name, age in result:print(name, age)

原生 SQL 查询：

SELECT name, age FROM users;

3.3 查询带条件的记录

查询年龄大于 30 的用户：

stmt = select(User).where(User.age > 30)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查询：

SELECT * FROM users WHERE age > 30;

3.4 排序查询

按年龄降序排列查询：

stmt = select(User).order_by(User.age.desc())with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查询：

SELECT * FROM users ORDER BY age DESC;

3.5 限制查询结果

限制返回前 5 条记录：

stmt = select(User).limit(5)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

原生 SQL 查询：

SELECT * FROM users LIMIT 5;

4. 关联查询（Join）

SQLAlchemy 提供了强大的关联查询功能。我们可以使用 joinedload​、subqueryload​ 等方法来优化查询。

4.1 懒查询 (Lazy Loading)

懒查询是 SQLAlchemy 默认的加载方式。只有当你访问某个关联属性时，SQLAlchemy 会延迟加载数据。

user = session.query(User).first()  # 执行查询，不查询 addresses
print(user.name)  # 输出用户名称
addresses = user.addresses  # 此时会查询 addresses 表
for address in addresses:print(address.email)

原生 SQL 查询：

SELECT * FROM users WHERE id = 1;
-- 查询地址
SELECT * FROM addresses WHERE user_id = 1;

4.2 预加载（Eager Loading）

为了避免 N+1 查询问题，可以使用预加载。以下是几种常用的预加载方式。

4.2.1 joinedload​

​joinedload​ 使用 SQL JOIN 一次性加载所有关联数据，适用于数据量较小的场景。

from sqlalchemy.orm import joinedloadstmt = select(User).options(joinedload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查询：

SELECT * FROM users
JOIN addresses ON users.id = addresses.user_id;
4.2.2 subqueryload​

​subqueryload​ 使用子查询加载关联数据，适用于关联数据较多的情况。

from sqlalchemy.orm import subqueryloadstmt = select(User).options(subqueryload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查询：

SELECT * FROM users;
-- 查询 addresses 表的所有数据
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);
4.2.3 selectinload​

​selectinload​ 通过批量查询一次性加载关联数据，适用于多对一和一对多的关系。

from sqlalchemy.orm import selectinloadstmt = select(User).options(selectinload(User.addresses))with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name)for address in user.addresses:print(address.email)

原生 SQL 查询：

SELECT * FROM users;
-- 查询 addresses 表
SELECT * FROM addresses WHERE user_id IN (1, 2, 3, ...);

4.3 延迟加载与预加载的比较

特性懒查询 (Lazy Loading)预加载 (Eager Loading)
数据加载方式延迟加载，直到访问关联属性时才查询一次性加载所有关联数据
查询次数可能多次查询（N+1 查询问题）一次性查询，通常只发出少量查询
性能对于小数据量高效对于复杂查询避免 N+1 问题，适用于大量关联数据
常用方法无​joinedload​、subqueryload​、selectinload​

5. 高级查询

5.1 聚合查询

SQLAlchemy 支持聚合函数，如 count​、sum​、avg​ 等。以下是计算用户数量的例子：

from sqlalchemy import funcstmt = select(func.count(User.id))with session.begin():result = session.execute(stmt).scalar()print(f"User Count: {result}")

SQL 原生查询：

SELECT COUNT(id) FROM users;

5.2 分组查询

通过 group_by​ 方法进行分组查询：

stmt = select(User.age, func.count(User.id)).group_by(User.age)with session.begin():result = session.execute(stmt).all()for age, count in result:print(f"Age: {age}, Count: {count}")

SQL 原生查询：

SELECT age, COUNT(id) FROM users GROUP BY age;

5.3 子查询

通过子查询来执行更复杂的查询：

subquery = select(func.max(User.age)).scalar_subquery()stmt = select(User).where(User.age == subquery)with session.begin():result = session.execute(stmt).scalars().all()for user in result:print(user.name, user.age)

SQL 原生查询：

SELECT * FROM users WHERE age = (SELECT MAX(age) FROM users);

6. 异步查询使用示例

SQLAlchemy 支持异步查询，适用于需要处理高并发操作的场景。以下是使用异步 session​ 进行查询的示例。

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select# 创建异步引擎和会话
engine = create_async_engine('sqlite+aiosqlite:///example.db', echo=True, future=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)async def async_query():async with async_session() as session:stmt = select(User)result = await session.execute(stmt)users = result.scalars().all()for user in users:print(user.name)# 执行异步查询
import asyncio
asyncio.run(async_query())

在异步查询中，我们使用 await​ 来执行查询，确保数据库操作不会阻塞主线程。

7. 总结

SQLAlchemy 2.0 提供了灵活且高效的查询功能，支持懒查询、预加载和异步操作。通过合理使用懒查询和预加载，可以有效地避免 N+1 查询问题，提高应用性能。在异步操作方面，SQLAlchemy 的异步会话使得我们可以在高并发环境下高效地进行数据库查询。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.pswp.cn/pingmian/82301.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Windows 使用 WSL 安装 Ubuntu

Windows 使用 WSL 安装 Ubuntu

一，前言 Windows 上轻松跑 Linux 又不想用笨重的VMware 和VirtualBox ，怎么办？ 开源项目 Windows Subsystem for Linux (WSL)。它解决了许多开发者在 Windows 和 Linux 间切换的痛点，实现在 Windows 上无缝跑 Linux 工具和命令。…
阅读更多...
[Excel VBA]如何製作買三送一優惠條件的POS結帳介面?

[Excel VBA]如何製作買三送一優惠條件的POS結帳介面?

Excel I VBA I 買三送一優惠條件的POS結帳機 因應商品特賣活動，結帳介面需整合特定優惠條件，如買三送一，買五送二等等優惠條件。本文封面影片以買三送一為範例，並搭配VBA和對應按鈕，而實現銷售訂單紀錄和即時更新庫存等…
阅读更多...
3d世界坐标系转屏幕坐标系

3d世界坐标系转屏幕坐标系

世界坐标 ——> NDC标准设备坐标 ——> 屏幕坐标 标准设备NDC坐标系 屏幕坐标系 .project方法将 将向量(坐标)从世界空间投影到相机的标准化设备坐标 (NDC) 空间。 手动实现HTML元素定位到模型位置，实现模型标签效果（和css2Render原理同理&#…
阅读更多...
Idea出现 100% classes 等

Idea出现 100% classes 等

总是误点出来，每次又忘了怎么消除，在这里记录一下。 出现这样： 操作idea界面的：点击View->Tool Windows ->Coverage，然后关掉
阅读更多...
从零开始学习QT——第一步

从零开始学习QT——第一步

一、Qt 1.1、Qt是什么 Qt是一个跨平台的C图形用户界面应用程序框架。它为应用程序开发者提供建立图形界面所需的所有功能。它是完全面向对象的，很容易扩展，并且允许真正的组件编程。 1.2、Qt的发展历程 1991年 Qt最早由芬兰奇趣科技开发 1996年 进入商业…
阅读更多...
MySQL 8.0 OCP 1Z0-908 171-180题

MySQL 8.0 OCP 1Z0-908 171-180题

Q171.Examine this MySQL client command to connect to a remote database: mysql-h remote-example.org-u root–protocolTCP–ssl-mode Which two–ss1-mode values will ensure that an X.509-compliant certificate will be used to establish the SSL/TLS connection to …
阅读更多...
【机器学习】 关于外插修正随机梯度方法的数值实验

【机器学习】 关于外插修正随机梯度方法的数值实验

1. 随机梯度下降（SGD） 迭代格式： x k 1 x k − η k ∇ f i ( x k ) x_{k1} x_k - \eta_k \nabla f_i(x_k) xk1​xk​−ηk​∇fi​(xk​) 其中， η k \eta_k ηk​ 为步长（可能递减）， ∇ f…
阅读更多...
每日c/c++题 备战蓝桥杯（洛谷P3382 三分法求极值详解）

每日c/c++题 备战蓝桥杯（洛谷P3382 三分法求极值详解）

洛谷P3382 三分法求极值详解 题目描述 P3382 三分法 要求在给定区间内寻找一个多项式函数的最大值点。题目保证函数在区间内先严格递增后严格递减（单峰函数），适合使用三分法求解。 算法原理 三分法核心思想 对于单峰函数，在区…
阅读更多...
[Windows] 一键实现重复工作自动化zTasker

[Windows] 一键实现重复工作自动化zTasker

zTasker，是一款定时｜热键｜纯粹的自动化任务神器。它支持超过100种任务类型，包括提醒、关机重启、报时、挡屏休息、文件备份、音量调节、静音等。用户可以通过定时、CPU占用、文件夹监控、网速、快捷键等多种条件触发任务。 简单点…
阅读更多...
Docker核心笔记

Docker核心笔记

一、概述 1、架构 Docker容器基于镜像运行，容器共享宿主机的内核，不会加载额外内核，通过Namespaces(环境隔离)和Cgroups(资源控制)实现隔离，Cgroups会限容器使用资源并控制优先级和统计数据。隔离后的容器仅包含应用所需的用户态依赖 2、安装 安装先卸载再安装，使用的yum…
阅读更多...
2025年电工杯数学建模B题【垃圾运输】原创论文分享

2025年电工杯数学建模B题【垃圾运输】原创论文分享

大家好呀，从发布赛题一直到现在，总算完成了2025年电工杯数学建模B题【垃圾运输】完整的成品论文。 给大家看一下目录吧： 目录 摘 要： 一、问题重述 二．问题分析 2.1问题一 2.2问题二 2.3问题三 三、模型假设 …
阅读更多...
[爬虫知识] IP代理

[爬虫知识] IP代理

相关实战案例：[爬虫实战] 代理爬取：小白也能看懂怎么用代理 相关爬虫专栏：JS逆向爬虫实战 爬虫知识点合集 爬虫实战案例 引言：爬虫与IP封锁的攻防战 对网络爬虫而言，遇到的一个较棘手的问题就是封IP：请…
阅读更多...
计算机视觉---YOLOv1

计算机视觉---YOLOv1

YOLOv1深度解析：单阶段目标检测的开山之作 一、YOLOv1概述 提出背景： 2016年由Joseph Redmon等人提出，全称"You Only Look Once"，首次将目标检测视为回归问题，开创单阶段（One-Stage&#xff09…
阅读更多...
前端学习笔记element-Plus

前端学习笔记element-Plus

【element-plus菜单】参数说明： active-text-color"#ffd04b"——激活颜色 background-color"#232323"——背景颜色（29,160,176） :default-active"$route.path"——配置默认高亮的菜单项 text-color"#f…
阅读更多...
【Django DRF】一篇文章总结Django DRF框架

【Django DRF】一篇文章总结Django DRF框架

第一章 DRF框架基础 1.1 DRF简介 1.1.1 DRF定义与作用 1. 定义 DRF 即 Django REST framework，它是一个建立在 Django 基础之上的强大且灵活的工具包，用于构建 Web API（应用程序编程接口）😎。简单来说，…
阅读更多...
如何解决 Python 项目安装依赖报错：ERROR: Failed to build installable wheels for some pyproject.toml based project

如何解决 Python 项目安装依赖报错：ERROR: Failed to build installable wheels for some pyproject.toml based project

如何解决 Python 项目安装依赖报错：ERROR: Failed to build installable wheels for some pyproject.toml based projects 在使用 pip 安装 Python 项目的依赖时，遇到类似如下的报错信息： ERROR: Failed to build installable wheels for s…
阅读更多...
使用f5-tts训练自己的模型笔记

使用f5-tts训练自己的模型笔记

摘要 服务器都有了，这不得练练丹，有点说不过去啊。所以尝试了从头开始训练一个模型，结果由于推理页面好像有bug，不知道是不是失败了，然后又尝试微调一下模型。本篇文章主要记录了三流调包侠尝试炼丹过程中学习到的一些…
阅读更多...
安全可控的AI底座：灯塔大模型应用开发平台全面实现国产信创兼容适配认证

安全可控的AI底座：灯塔大模型应用开发平台全面实现国产信创兼容适配认证

国产信创产品兼容适配认证是为了支持和推动国产信息技术产品和服务的发展而设立的一种质量标准和管理体系。适配认证旨在确保相关产品在安全性、可靠性、兼容性等方面达到一定的标准，以满足政府和关键行业对信息安全和自主可控的需求。 北京中烟创新科技有限公司&a…
阅读更多...
初识Vue【1】

初识Vue【1】

1.什么是Vue： Vue (读音 /vjuː/，类似于 **view**) 是一套用于构建用户界面的**渐进式框架**。与其它大型框架不同的是，Vue 被设计为可以自底向上逐层应用。Vue 的核心库只关注视图层，不仅易于上手，还便于与第三方库或…
阅读更多...
Jest入门

Jest入门

快速入门 Jest中文文档 | Jest中文网 1.下载：npm install --save-dev jest 2.创建 sum.js 文件： function sum(a, b) { return a b; } module.exports sum; 3.创建sum.test.js 的文件 const sum require(./sum); test(adds 1 2 to equal 3,…
阅读更多...
最新文章

Copyright @ 2022~2023