SQLAlchemy – 快速指南
SQLAlchemy – 快速指南
SQLAlchemy – 介绍
SQLAlchemy 是一个流行的 SQL 工具包和对象关系映射器。它是用Python编写的，为应用程序开发人员提供了 SQL 的全部功能和灵活性。它是在 MIT 许可下发布的开源和跨平台软件。
SQLAlchemy 以其对象关系映射器（ORM）而闻名，使用它可以将类映射到数据库，从而允许对象模型和数据库模式从一开始就以一种完全分离的方式进行开发。
随着 SQL 数据库的大小和性能开始变得重要，它们的行为不像对象集合。另一方面，随着对象集合中的抽象开始变得重要，它们的行为不像表和行。SQLAlchemy 旨在兼顾这两个原则。
出于这个原因，它采用了数据映射器模式（如 Hibernate），而不是许多其他 ORM 使用的活动记录模式。将使用 SQLAlchemy 从不同的角度查看数据库和 SQL。
Michael Bayer 是 SQLAlchemy 的原作者。其初始版本于 2006 年 2 月发布。最新版本编号为 1.2.7，最近发布于 2018 年 4 月。
什么是 ORM？
ORM（对象关系映射）是一种编程技术，用于在面向对象的编程语言中不兼容的类型系统之间转换数据。通常，在面向对象 (OO) 语言（如 Python）中使用的类型系统包含非标量类型。这些不能表示为原始类型，例如整数和字符串。因此，OO 程序员必须转换标量数据中的对象以与后端数据库交互。但是，大多数数据库产品（如 Oracle、MySQL 等）中的数据类型是主要的。
在 ORM 系统中，每个类都映射到底层数据库中的一个表。与其自己编写乏味的数据库接口代码，ORM 会为您处理这些问题，同时您可以专注于对系统逻辑进行编程。
SQLAlchemy – 环境设置
让我们讨论使用 SQLAlchemy 所需的环境设置。
安装 SQLAlchemy 需要任何高于 2.7 的 Python 版本。最简单的安装方法是使用 Python 包管理器pip。此实用程序与 Python 的标准发行版捆绑在一起。
pip install sqlalchemy
使用上述命令，我们可以从python.org下载最新发布的 SQLAlchemy版本并将其安装到您的系统中。
在 Python 的 anaconda 发行版的情况下，可以使用以下命令从conda 终端安装 SQLAlchemy –
conda install -c anaconda sqlalchemy
也可以从下面的源代码安装 SQLAlchemy –
python setup.py install
SQLAlchemy 旨在与为特定数据库构建的 DBAPI 实现一起操作。它使用方言系统与各种类型的 DBAPI 实现和数据库进行通信。所有方言都要求安装适当的 DBAPI 驱动程序。
以下是包括的方言 –
- 火鸟
- 微软 SQL 服务器
- MySQL
- 甲骨文
- PostgreSQL
- SQLite
- 赛贝斯
要检查 SQLAlchemy 是否已正确安装并了解其版本，请在 Python 提示符下输入以下命令 –
>>> import sqlalchemy >>>sqlalchemy.__version__ '1.2.7'
SQLAlchemy 核心 – 表达式语言
SQLAlchemy 核心包括SQL 渲染引擎、DBAPI 集成、事务集成和模式描述服务。SQLAlchemy 核心使用 SQL 表达式语言，该语言提供以模式为中心的使用范例，而 SQLAlchemy ORM 是一种以域为中心的使用模式。
SQL 表达式语言提供了一个使用 Python 构造表示关系数据库结构和表达式的系统。它提出了一种直接表示关系数据库的原始构造而无需意见的系统，这与 ORM 相比，ORM 提出了一种高级和抽象的使用模式，这本身就是表达式语言的应用用法示例。
表达式语言是 SQLAlchemy 的核心组件之一。它允许程序员在 Python 代码中指定 SQL 语句，并直接在更复杂的查询中使用它。表达式语言独立于后端，全面覆盖原始 SQL 的方方面面。它比 SQLAlchemy 中的任何其他组件更接近原始 SQL。
表达式语言直接表示关系数据库的原始结构。由于 ORM 基于表达式语言，典型的 Python 数据库应用程序可能会重叠使用两者。应用程序可以单独使用表达式语言，但它必须定义自己的系统来将应用程序概念转换为单独的数据库查询。
表达式语言的语句将被 SQLAlchemy 引擎翻译成相应的原始 SQL 查询。我们现在将学习如何创建引擎并在它的帮助下执行各种 SQL 查询。
SQLAlchemy 核心 – 连接到数据库
在上一章中，我们讨论了 SQLAlchemy 中的表达式语言。现在让我们继续执行连接到数据库所涉及的步骤。
Engine 类将Pool 和 Dialect连接在一起以提供数据库连接和行为的来源。使用create_engine()函数实例化 Engine 类的对象。
create_engine() 函数将数据库作为一个参数。不需要在任何地方定义数据库。标准调用形式必须将 URL 作为第一个位置参数发送，通常是指示数据库方言和连接参数的字符串。使用下面给出的代码，我们可以创建一个数据库。
>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///college.db', echo = True)
对于MySQL 数据库，请使用以下命令 –
engine = create_engine("mysql://user:pwd@localhost/college",echo = True)
要特别提及用于连接的DB-API，URL 字符串采用如下形式 –
dialect[+driver]://user:password@host/dbname
例如，如果您在MySQL中使用PyMySQL 驱动程序，请使用以下命令 –
mysql+pymysql://<username>:<password>@<host>/<dbname>
的回声标志是一个快捷方式设置SQLAlchemy的日志记录，其经由Python的标准日志记录模块来完成的。在后续章节中，我们将学习所有生成的 SQL。要隐藏详细输出，请将 echo 属性设置为None。create_engine() 函数的其他参数可能是特定于方言的。
create_engine() 函数返回一个Engine 对象。Engine 类的一些重要方法是 –
| Sr.No. | 方法和说明 |
|---|---|
| 1 |
connect() 返回连接对象 |
| 2 |
execute() 执行 SQL 语句构造 |
| 3 |
begin() 返回一个上下文管理器，通过建立的事务传递连接。操作成功后，事务提交，否则回滚 |
| 4 |
dispose() 处理引擎使用的连接池 |
| 5 |
driver() 引擎使用的方言的驱动程序名称 |
| 6 |
table_names() 返回数据库中所有可用表名的列表 |
| 7 |
transaction() 在事务边界内执行给定的函数 |
SQLAlchemy 核心 – 创建表
现在让我们讨论如何使用创建表功能。
SQL 表达式语言根据表列构造其表达式。SQLAlchemy Column 对象表示数据库表中的一列，而该列又由Tableobject表示。元数据包含表和关联对象（例如索引、视图、触发器等）的定义。
因此，来自 SQLAlchemy Metadata 的 MetaData 类的对象是 Table 对象及其关联模式构造的集合。它包含一组 Table 对象以及一个到 Engine 或 Connection 的可选绑定。
from sqlalchemy import MetaData meta = MetaData()
MetaData 类的构造函数可以具有默认为None 的绑定和架构参数。
接下来，我们使用类似于常规 SQL CREATE TABLE 语句的 Table 构造在上述元数据目录中定义所有表。
Table 类的对象表示数据库中的相应表。构造函数采用以下参数 –
| Name | 表名 |
|---|---|
| Metadata | 将保存此表的 MetaData 对象 |
| Column(s) | 列类的一个或多个对象 |
柱对象表示一个列在一个数据库表中。构造函数接受名称、类型和其他参数，例如primary_key、自动增量和其他约束。
SQLAlchemy 将 Python 数据与其中定义的最佳通用列数据类型相匹配。一些通用数据类型是 –
- 大整数
- 布尔值
- 日期
- 约会时间
- 漂浮
- 整数
- 数字
- 小整数
- 细绳
- 文本
- 时间
要在大学数据库中创建一个学生表，请使用以下代码段 –
from sqlalchemy import Table, Column, Integer, String, MetaData
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
create_all() 函数使用引擎对象创建所有定义的表对象并将信息存储在元数据中。
meta.create_all(engine)
下面给出了完整的代码，它将创建一个 SQLite 数据库 College.db，其中包含一个 Students 表。
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
meta.create_all(engine)
因为 create_engine() 函数的 echo 属性设置为True，控制台将显示表创建的实际 SQL 查询如下 –
CREATE TABLE students ( id INTEGER NOT NULL, name VARCHAR, lastname VARCHAR, PRIMARY KEY (id) )
College.db 将在当前工作目录中创建。要检查是否创建了学生表，您可以使用任何 SQLite GUI 工具（例如SQLiteStudio ）打开数据库。
下图显示了在数据库中创建的学生表 –
SQLAlchemy 核心 – SQL 表达式
在本章中，我们将简要介绍 SQL 表达式及其功能。
SQL 表达式是使用相对于目标表对象的相应方法构造的。例如，INSERT 语句是通过执行 insert() 方法创建的，如下所示 –
ins = students.insert()
上述方法的结果是一个插入对象，可以使用str()函数进行验证。下面的代码插入了学生 ID、姓名、姓氏等详细信息。
'INSERT INTO students (id, name, lastname) VALUES (:id, :name, :lastname)'
可以通过values()方法在特定字段中插入值以插入对象。下面给出了相同的代码 –
>>> ins = users.insert().values(name = 'Karan') >>> str(ins) 'INSERT INTO users (name) VALUES (:name)'
Python 控制台上回显的 SQL 不显示实际值（在本例中为“Karan”）。相反，SQLALchemy 生成一个绑定参数，该参数在语句的编译形式中可见。
ins.compile().params
{'name': 'Karan'}
类似地，update()、delete()和select() 等方法分别创建 UPDATE、DELETE 和 SELECT 表达式。我们将在后面的章节中了解它们。
SQLAlchemy 核心 – 执行表达式
在上一章中，我们学习了 SQL 表达式。在本章中，我们将研究这些表达式的执行。
为了执行生成的 SQL 表达式，我们必须获取一个表示主动检出的 DBAPI 连接资源的连接对象，然后提供表达式对象，如下面的代码所示。
conn = engine.connect()
以下 insert() 对象可用于 execute() 方法 –
ins = students.insert().values(name = 'Ravi', lastname = 'Kapoor') result = conn.execute(ins)
控制台显示 SQL 表达式的执行结果如下 –
INSERT INTO students (name, lastname) VALUES (?, ?)
('Ravi', 'Kapoor')
COMMIT
以下是显示使用 SQLAlchemy 的核心技术执行 INSERT 查询的整个片段 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
ins = students.insert()
ins = students.insert().values(name = 'Ravi', lastname = 'Kapoor')
conn = engine.connect()
result = conn.execute(ins)
结果可以通过使用 SQLite Studio 打开数据库来验证，如下面的屏幕截图所示 –
结果变量称为ResultProxy 对象。它类似于 DBAPI 游标对象。我们可以使用ResultProxy.inserted_primary_key获取有关从我们的语句生成的主键值的信息，如下所示 –
result.inserted_primary_key [1]
要使用 DBAPI 的 execute many() 方法发出多次插入，我们可以发送一个字典列表，每个字典包含一组不同的要插入的参数。
conn.execute(students.insert(), [
{'name':'Rajiv', 'lastname' : 'Khanna'},
{'name':'Komal','lastname' : 'Bhandari'},
{'name':'Abdul','lastname' : 'Sattar'},
{'name':'Priya','lastname' : 'Rajhans'},
])
这反映在表的数据视图中，如下图所示 –
SQLAlchemy 核心 – 选择行
在本章中，我们将讨论在表对象中选择行的概念。
表对象的 select() 方法使我们能够构造 SELECT 表达式。
s = students.select()
select 对象通过 str(s) 函数转换为SELECT 查询，如下所示 –
'SELECT students.id, students.name, students.lastname FROM students'
我们可以使用这个选择对象作为连接对象的 execute() 方法的参数，如下面的代码所示 –
result = conn.execute(s)
执行上述语句时，Python shell 会响应以下等效的 SQL 表达式 –
SELECT students.id, students.name, students.lastname FROM students
结果变量相当于 DBAPI 中的游标。我们现在可以使用fetchone() 方法获取记录。
row = result.fetchone()
表中所有选定的行都可以通过for 循环打印，如下所示 –
for row in result: print (row)
打印学生表中所有行的完整代码如下所示 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
s = students.select()
conn = engine.connect()
result = conn.execute(s)
for row in result:
print (row)
Python shell 中显示的输出如下 –
(1, 'Ravi', 'Kapoor') (2, 'Rajiv', 'Khanna') (3, 'Komal', 'Bhandari') (4, 'Abdul', 'Sattar') (5, 'Priya', 'Rajhans')
SELECT 查询的 WHERE 子句可以通过使用Select.where()来应用。例如，如果我们想显示 id >2 的行
s = students.select().where(students.c.id>2) result = conn.execute(s) for row in result: print (row)
这里c 属性是 column 的别名。以下输出将显示在外壳上 –
(3, 'Komal', 'Bhandari') (4, 'Abdul', 'Sattar') (5, 'Priya', 'Rajhans')
这里要注意，select对象也可以通过sqlalchemy.sql模块中的select()函数获取。select() 函数需要表对象作为参数。
from sqlalchemy.sql import select s = select([users]) result = conn.execute(s)
SQLAlchemy 核心 – 使用文本 SQL
SQLAlchemy 允许您只使用字符串，适用于 SQL 已知且不强烈需要语句支持动态功能的情况。text() 构造用于编写文本语句，该语句传递到数据库时基本保持不变。
它构造了一个新的TextClause，直接表示文本 SQL 字符串，如下面的代码所示 –
from sqlalchemy import text
t = text("SELECT * FROM students")
result = connection.execute(t)
text()比普通字符串提供的优点是 –
- 对绑定参数的后端中立支持
- 每个语句的执行选项
- 结果列类型行为
text() 函数需要命名冒号格式的绑定参数。无论数据库后端如何，它们都是一致的。要为参数发送值，我们将它们作为附加参数传递到 execute() 方法中。
以下示例在文本 SQL 中使用绑定参数 –
from sqlalchemy.sql import text
s = text("select students.name, students.lastname from students where students.name between :x and :y")
conn.execute(s, x = 'A', y = 'L').fetchall()
text() 函数构造 SQL 表达式如下 –
select students.name, students.lastname from students where students.name between ? and ?
x = ‘A’ 和 y = ‘L’ 的值作为参数传递。结果是名称在“A”和“L”之间的行列表 –
[('Komal', 'Bhandari'), ('Abdul', 'Sattar')]
text() 构造支持使用 TextClause.bindparams() 方法预先建立的绑定值。参数也可以显式输入如下 –
stmt = text("SELECT * FROM students WHERE students.name BETWEEN :x AND :y")
stmt = stmt.bindparams(
bindparam("x", type_= String),
bindparam("y", type_= String)
)
result = conn.execute(stmt, {"x": "A", "y": "L"})
The text() function also be produces fragments of SQL within a select() object that
accepts text() objects as an arguments. The “geometry” of the statement is provided by
select() construct , and the textual content by text() construct. We can build a statement
without the need to refer to any pre-established Table metadata.
from sqlalchemy.sql import select
s = select([text("students.name, students.lastname from students")]).where(text("students.name between :x and :y"))
conn.execute(s, x = 'A', y = 'L').fetchall()
您还可以使用and_()函数在 text() 函数的帮助下创建的 WHERE 子句中组合多个条件。
from sqlalchemy import and_
from sqlalchemy.sql import select
s = select([text("* from students")]) \
.where(
and_(
text("students.name between :x and :y"),
text("students.id>2")
)
)
conn.execute(s, x = 'A', y = 'L').fetchall()
上面的代码获取名称在“A”和“L”之间且 id 大于 2 的行。代码的输出如下 –
[(3, 'Komal', 'Bhandari'), (4, 'Abdul', 'Sattar')]
SQLAlchemy 核心 – 使用别名
SQL 中的别名对应于表或 SELECT 语句的“重命名”版本，只要您说“SELECT * FROM table1 AS a”，就会发生这种情况。AS 为表创建一个新名称。别名允许通过唯一名称引用任何表或子查询。
对于表，这允许在 FROM 子句中多次命名同一个表。它为语句所表示的列提供父名称，允许相对于该名称引用它们。
在 SQLAlchemy 中，任何 Table、select() 构造或其他可选对象都可以使用From Clause.alias()方法转换为别名，该方法生成别名构造。sqlalchemy.sql 模块中的 alias() 函数表示别名，通常应用于使用 AS 关键字的 SQL 语句中的任何表或子选择。
from sqlalchemy.sql import alias
st = students.alias("a")
现在可以在 select() 构造中使用此别名来引用学生表 –
s = select([st]).where(st.c.id>2)
这转换为 SQL 表达式如下 –
SELECT a.id, a.name, a.lastname FROM students AS a WHERE a.id > 2
我们现在可以使用连接对象的 execute() 方法执行这个 SQL 查询。完整的代码如下 –
from sqlalchemy.sql import alias, select
st = students.alias("a")
s = select([st]).where(st.c.id > 2)
conn.execute(s).fetchall()
当上面的代码行被执行时，它会生成以下输出 –
[(3, 'Komal', 'Bhandari'), (4, 'Abdul', 'Sattar'), (5, 'Priya', 'Rajhans')]
使用 UPDATE 表达式
目标表对象上的update()方法构造等效的 UPDATE SQL 表达式。
table.update().where(conditions).values(SET expressions)
的值（）所得到的更新对象的方法被用来指定更新的设定条件。如果保留为 None，则 SET 条件由在语句的执行和/或编译期间传递给语句的那些参数确定。
where 子句是一个可选表达式，用于描述 UPDATE 语句的 WHERE 条件。
以下代码片段将学生表中“姓氏”列的值从“Khanna”更改为“Kapoor” –
stmt = students.update().where(students.c.lastname == 'Khanna').values(lastname = 'Kapoor')
stmt 对象是一个更新对象，转换为 –
'UPDATE students SET lastname = :lastname WHERE students.lastname = :lastname_1'
当调用execute()方法时，绑定参数lastname_1将被替换。完整的更新代码如下 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
students = Table(
'students',
meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
conn = engine.connect()
stmt=students.update().where(students.c.lastname=='Khanna').values(lastname='Kapoor')
conn.execute(stmt)
s = students.select()
conn.execute(s).fetchall()
上面的代码显示以下输出，第二行显示更新操作的效果，如给出的屏幕截图所示 –
[ (1, 'Ravi', 'Kapoor'), (2, 'Rajiv', 'Kapoor'), (3, 'Komal', 'Bhandari'), (4, 'Abdul', 'Sattar'), (5, 'Priya', 'Rajhans') ]
请注意，也可以通过在 sqlalchemy.sql.expression 模块中使用update()函数来实现类似的功能，如下所示 –
from sqlalchemy.sql.expression import update stmt = update(students).where(students.c.lastname == 'Khanna').values(lastname = 'Kapoor')
使用 DELETE 表达式
在前一章中，我们已经了解了Update表达式的作用。我们要学习的下一个表达式是Delete。
删除操作可以通过在目标表对象上运行 delete() 方法来实现，如下面的语句所示 –
stmt = students.delete()
在学生表的情况下，上面的代码行构造一个 SQL 表达式如下 –
'DELETE FROM students'
但是，这将删除学生表中的所有行。通常 DELETE 查询与 WHERE 子句指定的逻辑表达式相关联。以下语句显示 where 参数 –
stmt = students.delete().where(students.c.id > 2)
生成的 SQL 表达式将有一个绑定参数，该参数将在运行时执行语句时被替换。
'DELETE FROM students WHERE students.id > :id_1'
以下代码示例将从学生表中删除姓氏为“Khanna”的行 –
from sqlalchemy.sql.expression import update
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
conn = engine.connect()
stmt = students.delete().where(students.c.lastname == 'Khanna')
conn.execute(stmt)
s = students.select()
conn.execute(s).fetchall()
要验证结果，请刷新 SQLiteStudio 中学生表的数据视图。
SQLAlchemy 核心 – 使用多个表
RDBMS 的重要特性之一是建立表之间的关系。可以对相关表执行 SELECT、UPDATE 和 DELETE 等 SQL 操作。本节使用 SQLAlchemy 描述这些操作。
为此，在我们的 SQLite 数据库 (college.db) 中创建了两个表。student 表的结构与上一节中给出的结构相同；而地址表具有st_id列，该列使用外键约束映射到学生表中的 id 列。
以下代码将在 College.db 中创建两个表 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
engine = create_engine('sqlite:///college.db', echo=True)
meta = MetaData()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
addresses = Table(
'addresses', meta,
Column('id', Integer, primary_key = True),
Column('st_id', Integer, ForeignKey('students.id')),
Column('postal_add', String),
Column('email_add', String))
meta.create_all(engine)
上面的代码将转换为学生和地址表的 CREATE TABLE 查询，如下所示 –
CREATE TABLE students ( id INTEGER NOT NULL, name VARCHAR, lastname VARCHAR, PRIMARY KEY (id) ) CREATE TABLE addresses ( id INTEGER NOT NULL, st_id INTEGER, postal_add VARCHAR, email_add VARCHAR, PRIMARY KEY (id), FOREIGN KEY(st_id) REFERENCES students (id) )
以下屏幕截图非常清楚地展示了上述代码 –
这些表通过执行表对象的insert() 方法填充数据。要在学生表中插入 5 行，您可以使用下面给出的代码 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
conn.execute(students.insert(), [
{'name':'Ravi', 'lastname':'Kapoor'},
{'name':'Rajiv', 'lastname' : 'Khanna'},
{'name':'Komal','lastname' : 'Bhandari'},
{'name':'Abdul','lastname' : 'Sattar'},
{'name':'Priya','lastname' : 'Rajhans'},
])
借助以下代码在地址表中添加行–
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
addresses = Table(
'addresses', meta,
Column('id', Integer, primary_key = True),
Column('st_id', Integer),
Column('postal_add', String),
Column('email_add', String)
)
conn.execute(addresses.insert(), [
{'st_id':1, 'postal_add':'Shivajinagar Pune', 'email_add':'[email protected]'},
{'st_id':1, 'postal_add':'ChurchGate Mumbai', 'email_add':'[email protected]'},
{'st_id':3, 'postal_add':'Jubilee Hills Hyderabad', 'email_add':'[email protected]'},
{'st_id':5, 'postal_add':'MG Road Bangaluru', 'email_add':'[email protected]'},
{'st_id':2, 'postal_add':'Cannought Place new Delhi', 'email_add':'[email protected]'},
])
请注意，addresses 表中的 st_id 列是指 Students 表中的 id 列。我们现在可以使用这个关系从两个表中获取数据。我们想从与addresses 表中的st_id 对应的students 表中获取name和lastname。
from sqlalchemy.sql import select s = select([students, addresses]).where(students.c.id == addresses.c.st_id) result = conn.execute(s) for row in result: print (row)
选择对象将有效地转换为以下 SQL 表达式，连接两个表的公共关系 –
SELECT students.id, students.name, students.lastname, addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM students, addresses WHERE students.id = addresses.st_id
这将产生从两个表中提取相应数据的输出，如下所示 –
(1, 'Ravi', 'Kapoor', 1, 1, 'Shivajinagar Pune', '[email protected]') (1, 'Ravi', 'Kapoor', 2, 1, 'ChurchGate Mumbai', '[email protected]') (3, 'Komal', 'Bhandari', 3, 3, 'Jubilee Hills Hyderabad', '[email protected]') (5, 'Priya', 'Rajhans', 4, 5, 'MG Road Bangaluru', '[email protected]') (2, 'Rajiv', 'Khanna', 5, 2, 'Cannought Place new Delhi', '[email protected]')
使用多个表更新
在上一章中，我们讨论了如何使用多个表。所以我们更进一步，在本章中学习多个表更新。
使用 SQLAlchemy 的 table 对象，可以在 update() 方法的 WHERE 子句中指定多个表。PostgreSQL 和 Microsoft SQL Server 支持引用多个表的 UPDATE 语句。这实现了“UPDATE FROM”语法，一次更新一张表。但是，可以直接在 WHERE 子句中的附加“FROM”子句中引用附加表。下面几行代码清楚地解释了多表更新的概念。
stmt = students.update().\
values({
students.c.name:'xyz',
addresses.c.email_add:'[email protected]'
}).\
where(students.c.id == addresses.c.id)
更新对象等效于以下 UPDATE 查询 –
UPDATE students SET email_add = :addresses_email_add, name = :name FROM addresses WHERE students.id = addresses.id
就 MySQL 方言而言，可以将多个表嵌入到由逗号分隔的单个 UPDATE 语句中，如下所示 –
stmt = students.update().\ values(name = 'xyz').\ where(students.c.id == addresses.c.id)
以下代码描述了生成的 UPDATE 查询 –
'UPDATE students SET name = :name FROM addresses WHERE students.id = addresses.id'
然而，SQLite 方言不支持 UPDATE 中的多表标准并显示以下错误 –
NotImplementedError: This backend does not support multiple-table criteria within UPDATE
参数顺序更新
原始 SQL 的 UPDATE 查询具有 SET 子句。它由 update() 构造使用原始 Table 对象中给出的列顺序呈现。因此，具有特定列的特定 UPDATE 语句每次将呈现相同。由于参数本身作为 Python 字典键传递给 Update.values() 方法，因此没有其他可用的固定顺序。
在某些情况下，SET 子句中呈现的参数顺序很重要。在 MySQL 中，对列值的更新基于其他列值的更新。
以下语句的结果 –
UPDATE table1 SET x = y + 10, y = 20
将产生与以下不同的结果 –
UPDATE table1 SET y = 20, x = y + 10
MySQL 中的 SET 子句是基于每个值而不是基于每行评估的。为此，使用了preserve_parameter_order。2 元组的 Python 列表作为Update.values()方法的参数给出–
stmt = table1.update(preserve_parameter_order = True).\ values([(table1.c.y, 20), (table1.c.x, table1.c.y + 10)])
List 对象类似于字典，除了它是有序的。这确保了“y”列的 SET 子句将首先呈现，然后是“x”列的 SET 子句。
SQLAlchemy 核心 – 多表删除
在本章中，我们将研究多表删除表达式，它类似于使用多表更新功能。
在许多 DBMS 方言中，DELETE 语句的 WHERE 子句中可以引用多个表。对于 PG 和 MySQL，使用“DELETE USING”语法；而对于 SQL Server，使用“DELETE FROM”表达式是指多个表。SQLAlchemy delete()构造通过在 WHERE 子句中指定多个表来隐式支持这两种模式，如下所示 –
stmt = users.delete().\
where(users.c.id == addresses.c.id).\
where(addresses.c.email_address.startswith('xyz%'))
conn.execute(stmt)
在 PostgreSQL 后端，上述语句的结果 SQL 将呈现为 –
DELETE FROM users USING addresses WHERE users.id = addresses.id AND (addresses.email_address LIKE %(email_address_1)s || '%%')
如果此方法与不支持此行为的数据库一起使用，编译器将引发 NotImplementedError。
SQLAlchemy 核心 – 使用连接
在本章中，我们将学习如何在 SQLAlchemy 中使用 Joins。
连接的效果是通过在 select() 构造的列子句或where 子句中放置两个表来实现的。现在我们使用 join() 和 outerjoin() 方法。
join() 方法返回一个从一个表对象到另一个表对象的连接对象。
join(right, onclause = None, isouter = False, full = False)
上面代码中提到的参数的功能如下 –
-
right – 连接的右侧；这是任何 Table 对象
-
onclause – 表示连接的 ON 子句的 SQL 表达式。如果保留为 None，则尝试根据外键关系连接两个表
-
isouter – 如果为 True，则呈现 LEFT OUTER JOIN，而不是 JOIN
-
full – 如果为 True，则呈现 FULL OUTER JOIN，而不是 LEFT OUTER JOIN
例如，以下使用 join() 方法将自动导致基于外键的连接。
>>> print(students.join(addresses))
这等效于以下 SQL 表达式 –
students JOIN addresses ON students.id = addresses.st_id
您可以明确提及加入标准如下 –
j = students.join(addresses, students.c.id == addresses.c.st_id)
如果我们现在使用此连接构建以下选择构造 –
stmt = select([students]).select_from(j)
这将导致以下 SQL 表达式 –
SELECT students.id, students.name, students.lastname FROM students JOIN addresses ON students.id = addresses.st_id
如果使用连接表示引擎执行此语句，则将显示属于所选列的数据。完整的代码如下 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
addresses = Table(
'addresses', meta,
Column('id', Integer, primary_key = True),
Column('st_id', Integer,ForeignKey('students.id')),
Column('postal_add', String),
Column('email_add', String)
)
from sqlalchemy import join
from sqlalchemy.sql import select
j = students.join(addresses, students.c.id == addresses.c.st_id)
stmt = select([students]).select_from(j)
result = conn.execute(stmt)
result.fetchall()
以下是上述代码的输出 –
[ (1, 'Ravi', 'Kapoor'), (1, 'Ravi', 'Kapoor'), (3, 'Komal', 'Bhandari'), (5, 'Priya', 'Rajhans'), (2, 'Rajiv', 'Khanna') ]
SQLAlchemy 核心 – 使用连接
连接是 SQLAlchemy 模块中的函数，用于实现 SQL 表达式的 WHERE 子句中使用的关系运算符。运算符 AND、OR、NOT 等用于形成组合两个单独逻辑表达式的复合表达式。在 SELECT 语句中使用 AND 的简单示例如下 –
SELECT * from EMPLOYEE WHERE salary>10000 AND age>30
SQLAlchemy 函数 and_()、or_() 和 not_() 分别实现 AND、OR 和 NOT 运算符。
and_() 函数
它产生由 AND 连接的表达式的连接。为了更好地理解，下面给出了一个例子 –
from sqlalchemy import and_
print(
and_(
students.c.name == 'Ravi',
students.c.id <3
)
)
这转化为 –
students.name = :name_1 AND students.id < :id_1
要在学生表的 select() 构造中使用 and_()，请使用以下代码行 –
stmt = select([students]).where(and_(students.c.name == 'Ravi', students.c.id <3))
将构造以下性质的 SELECT 语句 –
SELECT students.id, students.name, students.lastname FROM students WHERE students.name = :name_1 AND students.id < :id_1
显示上述 SELECT 查询输出的完整代码如下 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey, select
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
from sqlalchemy import and_, or_
stmt = select([students]).where(and_(students.c.name == 'Ravi', students.c.id <3))
result = conn.execute(stmt)
print (result.fetchall())
假设学生表中填充了上一个示例中使用的数据，将选择以下行 –
[(1, 'Ravi', 'Kapoor')]
or_() 函数
它产生由 OR 连接的表达式的连接。我们将使用 or_() 将上面示例中的 stmt 对象替换为以下对象
stmt = select([students]).where(or_(students.c.name == 'Ravi', students.c.id <3))
这将有效地等同于以下 SELECT 查询 –
SELECT students.id, students.name, students.lastname FROM students WHERE students.name = :name_1 OR students.id < :id_1
进行替换并运行上述代码后，结果将是落在 OR 条件中的两行 –
[(1, 'Ravi', 'Kapoor'), (2, 'Rajiv', 'Khanna')]
asc() 函数
它产生一个升序的 ORDER BY 子句。该函数将要应用该函数的列作为参数。
from sqlalchemy import asc stmt = select([students]).order_by(asc(students.c.name))
该语句实现以下 SQL 表达式 –
SELECT students.id, students.name, students.lastname FROM students ORDER BY students.name ASC
以下代码按姓名列的升序列出学生表中的所有记录 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey, select
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
students = Table(
'students', meta,
Column('id', Integer, primary_key = True),
Column('name', String),
Column('lastname', String),
)
from sqlalchemy import asc
stmt = select([students]).order_by(asc(students.c.name))
result = conn.execute(stmt)
for row in result:
print (row)
上面的代码产生以下输出 –
(4, 'Abdul', 'Sattar') (3, 'Komal', 'Bhandari') (5, 'Priya', 'Rajhans') (2, 'Rajiv', 'Khanna') (1, 'Ravi', 'Kapoor')
desc() 函数
类似的 desc() 函数产生如下降序 ORDER BY 子句 –
from sqlalchemy import desc stmt = select([students]).order_by(desc(students.c.lastname))
等效的 SQL 表达式是 –
SELECT students.id, students.name, students.lastname FROM students ORDER BY students.lastname DESC
以上代码行的输出是 –
(4, 'Abdul', 'Sattar') (5, 'Priya', 'Rajhans') (2, 'Rajiv', 'Khanna') (1, 'Ravi', 'Kapoor') (3, 'Komal', 'Bhandari')
之间（）函数
它产生一个 BETWEEN 谓词子句。这通常用于验证某个列的值是否在某个范围之间。例如，以下代码选择 id 列在 2 和 4 之间的行 –
from sqlalchemy import between stmt = select([students]).where(between(students.c.id,2,4)) print (stmt)
生成的 SQL 表达式类似于 –
SELECT students.id, students.name, students.lastname FROM students WHERE students.id BETWEEN :id_1 AND :id_2
结果如下 –
(2, 'Rajiv', 'Khanna') (3, 'Komal', 'Bhandari') (4, 'Abdul', 'Sattar')
SQLAlchemy 核心 – 使用函数
本章讨论了 SQLAlchemy 中使用的一些重要函数。
标准 SQL 推荐了许多由大多数方言实现的函数。它们根据传递给它的参数返回单个值。一些 SQL 函数将列作为参数，而一些是通用的。SQLAlchemy API 中的func 关键字用于生成这些函数。
在 SQL 中，now() 是一个泛型函数。以下语句使用 func 呈现 now() 函数 –
from sqlalchemy.sql import func result = conn.execute(select([func.now()])) print (result.fetchone())
上述代码的示例结果可能如下所示 –
(datetime.datetime(2018, 6, 16, 6, 4, 40),)
另一方面， count() 函数返回从表中选择的行数，通过使用 func 呈现 –
from sqlalchemy.sql import func result = conn.execute(select([func.count(students.c.id)])) print (result.fetchone())
从上面的代码中，将获取学生表中的行数。
使用 Employee 表和以下数据演示了一些内置 SQL 函数 –
| ID | 名称 | 分数 |
|---|---|---|
| 1 | 卡迈勒 | 56 |
| 2 | 费尔南德斯 | 85 |
| 3 | 苏尼尔 | 62 |
| 4 | 巴斯卡 | 76 |
max() 函数是通过使用 SQLAlchemy 中的 func 来实现的，这将导致 85，即获得的总最大分数 –
from sqlalchemy.sql import func result = conn.execute(select([func.max(employee.c.marks)])) print (result.fetchone())
类似地，将返回 56 个最小标记的 min() 函数将通过以下代码呈现 –
from sqlalchemy.sql import func result = conn.execute(select([func.min(employee.c.marks)])) print (result.fetchone())
因此，也可以使用以下代码来实现 AVG() 函数 –
from sqlalchemy.sql import func
result = conn.execute(select([func.avg(employee.c.marks)]))
print (result.fetchone())
Functions are normally used in the columns clause of a select statement.
They can also be given label as well as a type. A label to function allows the result
to be targeted in a result row based on a string name, and a type is required when
you need result-set processing to occur.from sqlalchemy.sql import func
result = conn.execute(select([func.max(students.c.lastname).label('Name')]))
print (result.fetchone())
SQLAlchemy 核心 – 使用集合操作
在上一章中，我们学习了 max()、min()、count() 等各种函数，在这里，我们将学习集合操作及其用途。
标准 SQL 及其大部分方言支持 UNION 和 INTERSECT 等集合操作。SQLAlchemy 在以下函数的帮助下实现它们 –
联盟（）
在组合两个或多个 SELECT 语句的结果时，UNION 从结果集中消除重复项。两个表中的列数和数据类型必须相同。
union() 函数从多个表中返回一个 CompoundSelect 对象。以下示例演示了它的用法 –
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, union
engine = create_engine('sqlite:///college.db', echo = True)
meta = MetaData()
conn = engine.connect()
addresses = Table(
'addresses', meta,
Column('id', Integer, primary_key = True),
Column('st_id', Integer),
Column('postal_add', String),
Column('email_add', String)
)
u = union(addresses.select().where(addresses.c.email_add.like('%@gmail.com addresses.select().where(addresses.c.email_add.like('%@yahoo.com'))))
result = conn.execute(u)
result.fetchall()
联合构造转换为以下 SQL 表达式 –
SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ? UNION SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ?
从我们的地址表中，以下几行代表联合操作 –
[ (1, 1, 'Shivajinagar Pune', '[email protected]'), (2, 1, 'ChurchGate Mumbai', '[email protected]'), (3, 3, 'Jubilee Hills Hyderabad', '[email protected]'), (4, 5, 'MG Road Bangaluru', '[email protected]') ]
union_all()
UNION ALL 操作无法删除重复项，也无法对结果集中的数据进行排序。例如，在上面的查询中，将 UNION 替换为 UNION ALL 以查看效果。
u = union_all(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.email_add.like('%@yahoo.com')))
相应的 SQL 表达式如下 –
SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ? UNION ALL SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ?
除了_（）
SQL EXCEPT子句/运算符用于组合两个 SELECT 语句并从第一个 SELECT 语句返回第二个 SELECT 语句未返回的行。except_() 函数生成一个带有 EXCEPT 子句的 SELECT 表达式。
在以下示例中，except_() 函数仅返回地址表中在 email_add 字段中包含 ‘gmail.com’ 的记录，但排除在 postal_add 字段中包含 ‘Pune’ 的记录。
u = except_(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.postal_add.like('%Pune')))
上述代码的结果是以下 SQL 表达式 –
SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ? EXCEPT SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.postal_add LIKE ?
假设地址表包含前面示例中使用的数据，它将显示以下输出 –
[(2, 1, 'ChurchGate Mumbai', '[email protected]'), (3, 3, 'Jubilee Hills Hyderabad', '[email protected]')]
相交（）
使用 INTERSECT 运算符，SQL 显示来自两个 SELECT 语句的公共行。intersect() 函数实现了这种行为。
在以下示例中，两个 SELECT 构造是 intersect() 函数的参数。一个返回包含 ‘gmail.com’ 作为 email_add 列的一部分的行，其他返回包含 ‘Pune’ 作为 postal_add 列一部分的行。结果将是来自两个结果集中的公共行。
u = intersect(addresses.select().where(addresses.c.email_add.like('%@gmail.com')), addresses.select().where(addresses.c.postal_add.like('%Pune')))
实际上，这等效于以下 SQL 语句 –
SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.email_add LIKE ? INTERSECT SELECT addresses.id, addresses.st_id, addresses.postal_add, addresses.email_add FROM addresses WHERE addresses.postal_add LIKE ?
两个绑定参数 ‘%gmail.com’ 和 ‘%Pune’ 从地址表中的原始数据生成一行，如下所示 –
[(1, 1, 'Shivajinagar Pune', '[email protected]')]
SQLAlchemy ORM – 声明映射
SQLAlchemy 的对象关系映射器 API 的主要目标是促进将用户定义的 Python 类与数据库表以及这些类的对象与其对应表中的行相关联。对象和行的状态变化相互同步匹配。SQLAlchemy 能够根据用户定义的类及其定义的关系来表达数据库查询。
ORM 构建在 SQL 表达式语言之上。它是一种高级和抽象的使用模式。实际上，ORM 是表达式语言的应用用法。
尽管可以专门使用对象关系映射器构建成功的应用程序，但有时使用 ORM 构建的应用程序可能会在需要特定数据库交互的情况下直接使用表达式语言。
声明映射
首先，调用 create_engine() 函数设置一个引擎对象，该对象随后用于执行 SQL 操作。该函数有两个参数，一个是数据库的名称，另一个是当设置为 True 时将生成活动日志的 echo 参数。如果它不存在，将创建数据库。在以下示例中，创建了一个 SQLite 数据库。
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
当调用 Engine.execute() 或 Engine.connect() 等方法时，引擎会建立到数据库的真实 DBAPI 连接。然后用于发出不直接使用引擎的 SQLORM；相反，它由 ORM 在幕后使用。
在 ORM 的情况下，配置过程首先描述数据库表，然后定义将映射到这些表的类。在 SQLAlchemy 中，这两个任务是一起执行的。这是通过使用声明式系统完成的；创建的类包括用于描述它们映射到的实际数据库表的指令。
基类在声明式系统中存储类和映射表的目录。这称为声明性基类。在通常导入的模块中通常只有一个此基础的实例。declarative_base() 函数用于创建基类。该函数在 sqlalchemy.ext.declarative 模块中定义。
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base()
一旦声明了基类，就可以根据它定义任意数量的映射类。以下代码定义了一个 Customer 的类。它包含要映射到的表，以及其中列的名称和数据类型。
class Customers(Base): __tablename__ = 'customers' id = Column(Integer, primary_key = True) name = Column(String) address = Column(String) email = Column(String)
Declarative 中的类必须具有__tablename__属性，并且至少有一个Column作为主键的一部分。声明性用称为描述符的特殊 Python 访问器替换所有Column对象。这个过程被称为检测，它提供了在 SQL 上下文中引用表的方法，并支持从数据库中持久化和加载列的值。
这个映射类就像一个普通的 Python 类一样，具有符合要求的属性和方法。
声明式系统中关于类的信息，称为表元数据。SQLAlchemy 使用 Table 对象来表示由 Declarative 创建的特定表的此信息。Table对象按照规范创建，通过构造Mapper对象与类关联。这个映射器对象不是直接使用的，而是在内部用作映射类和表之间的接口。
每个 Table 对象都是称为 MetaData 的更大集合的成员，并且可以使用声明性基类的.metadata属性访问该对象。所述MetaData.create_all（）方法是，在我们的发动机传递的数据库连接的源极。对于尚未创建的所有表，它会向数据库发出 CREATE TABLE 语句。
Base.metadata.create_all(engine)
创建数据库和表以及映射 Python 类的完整脚本如下 –
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Customers(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String)
address = Column(String)
email = Column(String)
Base.metadata.create_all(engine)
执行时，Python 控制台将在执行 SQL 表达式后回显 –
CREATE TABLE customers ( id INTEGER NOT NULL, name VARCHAR, address VARCHAR, email VARCHAR, PRIMARY KEY (id) )
如果我们使用 SQLiteStudio 图形工具打开 Sales.db，它会以上述结构显示其中的客户表。
SQLAlchemy ORM – 创建会话
为了与数据库交互，我们需要获取它的句柄。会话对象是数据库的句柄。会话类是使用 sessionmaker() 定义的——一个可配置的会话工厂方法，它绑定到之前创建的引擎对象。
from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind = engine)
然后使用其默认构造函数设置会话对象，如下所示 –
session = Session()
下面列出了一些经常需要的会话类方法 –
| Sr.No. | 方法和说明 |
|---|---|
| 1 |
begin() 在此会话上开始交易 |
| 2 |
add() 在会话中放置一个对象。它的状态在下次刷新操作时保留在数据库中 |
| 3 |
add_all() 向会话添加对象集合 |
| 4 |
commit() 刷新所有项目和任何正在进行的交易 |
| 5 |
delete() 将交易标记为已删除 |
| 6 |
execute() 执行一个 SQL 表达式 |
| 7 |
expire() 将实例的属性标记为过时 |
| 8 |
flush() 将所有对象更改刷新到数据库 |
| 9 |
invalidate() 使用连接失效关闭会话 |
| 10 |
rollback() 回滚当前正在进行的事务 |
| 11 |
close() 通过清除所有项目并结束正在进行的任何交易来关闭当前会话 |
SQLAlchemy ORM – 添加对象
在 SQLAlchemy ORM 的前几章中，我们学习了如何声明映射和创建会话。在本章中，我们将学习如何向表中添加对象。
我们已经声明了已经映射到客户表的 Customer 类。我们必须声明这个类的一个对象，并通过会话对象的 add() 方法将它持久地添加到表中。
c1 = Sales(name = 'Ravi Kumar', address = 'Station Road Nanded', email = '[email protected]') session.add(c1)
请注意，此事务处于挂起状态，直到使用 commit() 方法刷新它为止。
session.commit()
以下是在客户表中添加记录的完整脚本 –
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Customers(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String)
address = Column(String)
email = Column(String)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
c1 = Customers(name = 'Ravi Kumar', address = 'Station Road Nanded', email = '[email protected]')
session.add(c1)
session.commit()
要添加多条记录，我们可以使用会话类的add_all()方法。
session.add_all([ Customers(name = 'Komal Pande', address = 'Koti, Hyderabad', email = '[email protected]'), Customers(name = 'Rajender Nath', address = 'Sector 40, Gurgaon', email = '[email protected]'), Customers(name = 'S.M.Krishna', address = 'Budhwar Peth, Pune', email = '[email protected]')] ) session.commit()
SQLiteStudio 的表视图显示记录被持久地添加到客户表中。下图显示了结果 –
SQLAlchemy ORM – 使用查询
SQLAlchemy ORM 生成的所有 SELECT 语句都是由 Query 对象构造的。它提供了一个生成接口，因此连续调用会返回一个新的 Query 对象，即前者的副本，并带有附加条件和与之相关的选项。
查询对象最初是使用 Session 的 query() 方法生成的，如下所示 –
q = session.query(mapped class)
以下语句也等效于上面给出的语句 –
q = Query(mappedClass, session)
查询对象具有 all() 方法，该方法以对象列表的形式返回结果集。如果我们在客户表上执行它 –
result = session.query(Customers).all()
此语句实际上等效于以下 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers
可以使用 For 循环遍历结果对象，如下所示，以获取基础客户表中的所有记录。这是在客户表中显示所有记录的完整代码 –
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Customers(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key = True)
name = Column(String)
address = Column(String)
email = Column(String)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
result = session.query(Customers).all()
for row in result:
print ("Name: ",row.name, "Address:",row.address, "Email:",row.email)
Python 控制台显示如下记录列表 –
Name: Ravi Kumar Address: Station Road Nanded Email: [email protected] Name: Komal Pande Address: Koti, Hyderabad Email: [email protected] Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected] Name: S.M.Krishna Address: Budhwar Peth, Pune Email: [email protected]
Query 对象还有以下有用的方法 –
| Sr.No. | 方法和说明 |
|---|---|
| 1 |
add_columns() 它将一个或多个列表达式添加到要返回的结果列列表中。 |
| 2 |
add_entity() 它将映射实体添加到要返回的结果列列表中。 |
| 3 |
count() 它返回此查询将返回的行数。 |
| 4 |
delete() 它执行批量删除查询。从数据库中删除与此查询匹配的行。 |
| 5 |
distinct() 它将 DISTINCT 子句应用于查询并返回新生成的查询。 |
| 6 |
filter() 它使用 SQL 表达式将给定的过滤条件应用于此查询的副本。 |
| 7 |
first() 如果结果不包含任何行，则返回此 Query 的第一个结果或 None 。 |
| 8 |
get() 它根据给定的主键标识符返回一个实例，提供对拥有会话的身份映射的直接访问。 |
| 9 |
group_by() 它将一个或多个 GROUP BY 条件应用于查询并返回新生成的查询 |
| 10 |
join() 它根据此 Query 对象的标准创建 SQL JOIN 并生成应用，返回新生成的 Query。 |
| 11 |
one() 它只返回一个结果或引发异常。 |
| 12 |
order_by() 它将一个或多个 ORDER BY 条件应用于查询并返回新生成的查询。 |
| 13 |
update() 它执行批量更新查询并更新数据库中与此查询匹配的行。 |
SQLAlchemy ORM – 更新对象
在本章中，我们将看到如何使用所需值修改或更新表。
要修改任何对象的某个属性的数据，我们必须为其分配新值并提交更改以使更改持久化。
让我们从 ID=2 的客户表中主键标识符的表中获取一个对象。我们可以使用会话的 get() 方法如下 –
x = session.query(Customers).get(2)
我们可以使用以下给定的代码显示所选对象的内容 –
print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)
从我们的客户表中，应显示以下输出 –
Name: Komal Pande Address: Koti, Hyderabad Email: [email protected]
现在我们需要通过分配新值来更新地址字段，如下所示 –
x.address = 'Banjara Hills Secunderabad' session.commit()
更改将持久反映在数据库中。现在我们使用first() 方法获取与表中第一行对应的对象，如下所示 –
x = session.query(Customers).first()
这将执行以下 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers LIMIT ? OFFSET ?
绑定参数将分别为 LIMIT = 1 和 OFFSET = 0，这意味着将选择第一行。
print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)
现在，显示第一行的上述代码的输出如下 –
Name: Ravi Kumar Address: Station Road Nanded Email: [email protected]
现在更改名称属性并使用以下代码显示内容 –
x.name = 'Ravi Shrivastava'
print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)
上面代码的输出是 –
Name: Ravi Shrivastava Address: Station Road Nanded Email: [email protected]
即使显示更改，它也不会提交。您可以通过使用带有以下代码的rollback() 方法来保留较早的持久位置。
session.rollback()
print ("Name: ", x.name, "Address:", x.address, "Email:", x.email)
将显示第一条记录的原始内容。
对于批量更新，我们将使用 Query 对象的 update() 方法。让我们试着给一个前缀，“先生” 在每一行中命名（ID = 2 除外）。相应的 update() 语句如下 –
session.query(Customers).filter(Customers.id! = 2).
update({Customers.name:"Mr."+Customers.name}, synchronize_session = False)
update() 方法需要两个参数，如下所示 –
-
一个键值字典，键是要更新的属性，值是属性的新内容。
-
同步会话属性提到更新会话中的属性的策略。有效值为false：对于不同步会话，fetch：在更新前执行选择查询以查找更新查询匹配的对象；和评估：评估会话中对象的标准。
表中 4 行中的 3 行将名称以“先生”为前缀。但是，更改未提交，因此不会反映在 SQLiteStudio 的表视图中。只有当我们提交会话时它才会刷新。
SQLAlchemy ORM – 应用过滤器
在本章中，我们将讨论如何应用过滤器以及某些过滤器操作及其代码。
Query 对象表示的结果集可以通过使用 filter() 方法服从某些条件。过滤方法的一般用法如下 –
session.query(class).filter(criteria)
在以下示例中，通过对客户表的 SELECT 查询获得的结果集按条件过滤，(ID>2) –
result = session.query(Customers).filter(Customers.id>2)
此语句将转换为以下 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id > ?
由于绑定参数 (?) 为 2，因此只会显示 ID column>2 的那些行。完整代码如下 –
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Customers(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key = True)
name = Column(String)
address = Column(String)
email = Column(String)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
result = session.query(Customers).filter(Customers.id>2)
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
Python 控制台中显示的输出如下 –
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected] ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: [email protected]
SQLAlchemy ORM – 过滤操作符
现在，我们将学习过滤器操作及其各自的代码和输出。
等于
通常使用的运算符是 ==，它应用标准来检查相等性。
result = session.query(Customers).filter(Customers.id == 2)
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
SQLAlchemy 将发送以下 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id = ?
上述代码的输出如下 –
ID: 2 Name: Komal Pande Address: Banjara Hills Secunderabad Email: [email protected]
不等于
用于不等于的运算符是 != 并且它提供不等于条件。
result = session.query(Customers).filter(Customers.id! = 2)
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
生成的 SQL 表达式是 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id != ?
上述代码行的输出如下 –
ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: [email protected] ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected] ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: [email protected]
喜欢
like() 方法本身为 SELECT 表达式中的 WHERE 子句生成 LIKE 条件。
result = session.query(Customers).filter(Customers.name.like('Ra%'))
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
以上 SQLAlchemy 代码等效于以下 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.name LIKE ?
上面代码的输出是 –
ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: [email protected] ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected]
在
此运算符检查列值是否属于列表中的项目集合。它由 in_() 方法提供。
result = session.query(Customers).filter(Customers.id.in_([1,3]))
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
在这里，由 SQLite 引擎评估的 SQL 表达式如下：
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id IN (?, ?)
上述代码的输出如下 –
ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: [email protected] ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected]
和
这种连接是通过在过滤器中放置多个逗号分隔的条件或使用 and_() 方法生成的，如下所示 –
result = session.query(Customers).filter(Customers.id>2, Customers.name.like('Ra%'))
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
from sqlalchemy import and_
result = session.query(Customers).filter(and_(Customers.id>2, Customers.name.like('Ra%')))
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
上述两种方法都会产生类似的 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id > ? AND customers.name LIKE ?
以上代码行的输出是 –
ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected]
或者
这个连接是由or_() 方法实现的。
from sqlalchemy import or_
result = session.query(Customers).filter(or_(Customers.id>2, Customers.name.like('Ra%')))
for row in result:
print ("ID:", row.id, "Name: ",row.name, "Address:",row.address, "Email:",row.email)
因此，SQLite 引擎遵循等效的 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id > ? OR customers.name LIKE ?
上述代码的输出如下 –
ID: 1 Name: Ravi Kumar Address: Station Road Nanded Email: [email protected] ID: 3 Name: Rajender Nath Address: Sector 40, Gurgaon Email: [email protected] ID: 4 Name: S.M.Krishna Address: Budhwar Peth, Pune Email: [email protected]
返回列表和标量
有许多 Query 对象的方法可以立即发出 SQL 并返回一个包含加载的数据库结果的值。
这是返回列表和标量的简要概述 –
全部（）
它返回一个列表。下面给出的是 all() 函数的代码行。
session.query(Customers).all()
Python 控制台显示以下发出的 SQL 表达式 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers
第一的（）
它应用 1 的限制并将第一个结果作为标量返回。
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers LIMIT ? OFFSET ?
LIMIT 的边界参数为 1，OFFSET 的边界参数为 0。
一（）
此命令完全获取所有行，如果结果中不存在完全一个对象标识或复合行，则会引发错误。
session.query(Customers).one()
找到多行 –
MultipleResultsFound: Multiple rows were found for one()
没有找到行 –
NoResultFound: No row was found for one()
one() 方法对于期望以不同方式处理“未找到项目”和“找到多个项目”的系统很有用。
标量（）
它调用 one() 方法，成功后返回行的第一列，如下所示 –
session.query(Customers).filter(Customers.id == 3).scalar()
这会生成以下 SQL 语句 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE customers.id = ?
SQLAlchemy ORM – 文本 SQL
前面已经从SQLAlchemy的核心表达语言的角度解释了使用text()函数的文本SQL。现在我们将从 ORM 的角度讨论它。
通过使用 text() 构造指定文字字符串的使用，可以灵活地将文字字符串与 Query 对象一起使用。大多数适用的方法都接受它。例如，filter() 和 order_by()。
在下面给出的示例中，filter() 方法将字符串“id<3”转换为 WHERE id<3
from sqlalchemy import text
for cust in session.query(Customers).filter(text("id<3")):
print(cust.name)
生成的原始 SQL 表达式显示了过滤器到 WHERE 子句的转换，代码如下所示 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE id<3
从我们在客户表中的示例数据中，将选择两行，名称列将打印如下 –
Ravi Kumar Komal Pande
要使用基于字符串的 SQL 指定绑定参数，请使用冒号，并使用 params() 方法指定值。
cust = session.query(Customers).filter(text("id = :value")).params(value = 1).one()
Python 控制台上显示的有效 SQL 如下所示 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE id = ?
要使用完全基于字符串的语句，可以将表示完整语句的 text() 构造传递给 from_statement()。
session.query(Customers).from_statement(text("SELECT * FROM customers")).all()
上面代码的结果将是一个基本的 SELECT 语句，如下所示 –
SELECT * FROM customers
显然，customers 表中的所有记录都会被选中。
text() 构造允许我们在位置上将其文本 SQL 链接到 Core 或 ORM 映射的列表达式。我们可以通过将列表达式作为位置参数传递给 TextClause.columns() 方法来实现这一点。
stmt = text("SELECT name, id, name, address, email FROM customers")
stmt = stmt.columns(Customers.id, Customers.name)
session.query(Customers.id, Customers.name).from_statement(stmt).all()
即使 SQLite 引擎执行上述代码生成的以下表达式，所有行的 id 和 name 列也将被选择，显示 text() 方法中的所有列 –
SELECT name, id, name, address, email FROM customers
SQLAlchemy ORM – 建立关系
本次会议描述了另一个与我们数据库中现有表相关的表的创建。客户表包含客户的主数据。我们现在需要创建发票表，其中可能包含任意数量的属于客户的发票。这是一对多关系的情况。
使用声明式，我们定义这个表及其映射类，发票如下所示 –
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key = True)
name = Column(String)
address = Column(String)
email = Column(String)
class Invoice(Base):
__tablename__ = 'invoices'
id = Column(Integer, primary_key = True)
custid = Column(Integer, ForeignKey('customers.id'))
invno = Column(Integer)
amount = Column(Integer)
customer = relationship("Customer", back_populates = "invoices")
Customer.invoices = relationship("Invoice", order_by = Invoice.id, back_populates = "customer")
Base.metadata.create_all(engine)
这将向 SQLite 引擎发送一个 CREATE TABLE 查询，如下所示 –
CREATE TABLE invoices ( id INTEGER NOT NULL, custid INTEGER, invno INTEGER, amount INTEGER, PRIMARY KEY (id), FOREIGN KEY(custid) REFERENCES customers (id) )
我们可以在 SQLiteStudio 工具的帮助下检查 sales.db 中是否创建了新表。
发票类在 custid 属性上应用 ForeignKey 构造。该指令指示此列中的值应限制为客户表中 id 列中存在的值。这是关系型数据库的一个核心特性，是将未连接的表集合转化为具有丰富重叠关系的“粘合剂”。
第二个指令，称为relationship()，告诉ORM Invoice 类应该使用属性Invoice.customer 链接到Customer 类。关系（）使用两个表之间的外键关系来确定这种链接的性质，确定它是多对一的。
附加的relationship() 指令放置在Customer.invoices 属性下的Customer 映射类上。参数relationship.back_populates 被分配来引用互补的属性名称，以便每个relationship() 可以对反向表达的相同关系做出智能决策。一方面，Invoices.customer 指的是 Invoices 实例，另一方面，Customer.invoices 指的是一系列 Customer 实例。
关系函数是 SQLAlchemy ORM 包的关系 API 的一部分。它提供了两个映射类之间的关系。这对应于父子或关联表关系。
以下是发现的基本关系模式 –
一对多
一对多关系在子表上的外键的帮助下引用父级。然后在父级上指定关系（），作为引用由子级表示的项目集合。relationship.back_populates 参数用于建立一对多的双向关系，其中“反向”侧是多对一。
多对一
另一方面，多对一关系在父表中放置一个外键来引用子表。关系（）在父级上声明，其中将创建一个新的标量保持属性。在这里，relationship.back_populates 参数再次用于双向行为。
一对一
一对一关系本质上是一种双向关系。uselist 标志指示在关系的“多”端放置标量属性而不是集合。要将一对多转换为一对一类型的关系，请将 uselist 参数设置为 false。
多对多
多对多关系是通过添加与两个类相关的关联表来建立的，该关联表通过定义具有外键的属性来建立。它由关系（）的第二个参数指示。通常，Table 使用与声明性基类关联的 MetaData 对象，以便 ForeignKey 指令可以定位要链接的远程表。每个relationship() 的relationship.back_populates 参数建立双向关系。关系的双方都包含一个集合。
使用相关对象
在本章中，我们将重点介绍 SQLAlchemy ORM 中的相关对象。
现在当我们创建一个 Customer 对象时，一个空白的发票集合将以 Python 列表的形式出现。
c1 = Customer(name = "Gopal Krishna", address = "Bank Street Hydarebad", email = "[email protected]")
c1.invoices 的 invoices 属性将是一个空列表。我们可以将列表中的项目分配为 –
c1.invoices = [Invoice(invno = 10, amount = 15000), Invoice(invno = 14, amount = 3850)]
让我们使用 Session 对象将此对象提交到数据库，如下所示 –
from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind = engine) session = Session() session.add(c1) session.commit()
这将自动为客户和发票表生成 INSERT 查询 –
INSERT INTO customers (name, address, email) VALUES (?, ?, ?)
('Gopal Krishna', 'Bank Street Hydarebad', '[email protected]')
INSERT INTO invoices (custid, invno, amount) VALUES (?, ?, ?)
(2, 10, 15000)
INSERT INTO invoices (custid, invno, amount) VALUES (?, ?, ?)
(2, 14, 3850)
现在让我们在 SQLiteStudio 的表视图中查看客户表和发票表的内容 –
您可以使用以下命令通过在构造函数本身中提供发票的映射属性来构造 Customer 对象 –
c2 = [
Customer(
name = "Govind Pant",
address = "Gulmandi Aurangabad",
email = "[email protected]",
invoices = [Invoice(invno = 3, amount = 10000),
Invoice(invno = 4, amount = 5000)]
)
]
或者使用会话对象的 add_all() 函数添加的对象列表，如下所示 –
rows = [
Customer(
name = "Govind Kala",
address = "Gulmandi Aurangabad",
email = "[email protected]",
invoices = [Invoice(invno = 7, amount = 12000), Invoice(invno = 8, amount = 18500)]),
Customer(
name = "Abdul Rahman",
address = "Rohtak",
email = "[email protected]",
invoices = [Invoice(invno = 9, amount = 15000),
Invoice(invno = 11, amount = 6000)
])
]
session.add_all(rows)
session.commit()
SQLAlchemy ORM – 使用连接
现在我们有两个表，我们将看到如何同时在两个表上创建查询。为了在 Customer 和 Invoice 之间构建一个简单的隐式连接，我们可以使用 Query.filter() 将它们的相关列等同起来。下面，我们使用此方法一次加载 Customer 和 Invoice 实体 –
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
for c, i in session.query(Customer, Invoice).filter(Customer.id == Invoice.custid).all():
print ("ID: {} Name: {} Invoice No: {} Amount: {}".format(c.id,c.name, i.invno, i.amount))
SQLAlchemy 发出的 SQL 表达式如下 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email, invoices.id AS invoices_id, invoices.custid AS invoices_custid, invoices.invno AS invoices_invno, invoices.amount AS invoices_amount FROM customers, invoices WHERE customers.id = invoices.custid
以上代码行的结果如下 –
ID: 2 Name: Gopal Krishna Invoice No: 10 Amount: 15000 ID: 2 Name: Gopal Krishna Invoice No: 14 Amount: 3850 ID: 3 Name: Govind Pant Invoice No: 3 Amount: 10000 ID: 3 Name: Govind Pant Invoice No: 4 Amount: 5000 ID: 4 Name: Govind Kala Invoice No: 7 Amount: 12000 ID: 4 Name: Govind Kala Invoice No: 8 Amount: 8500 ID: 5 Name: Abdul Rahman Invoice No: 9 Amount: 15000 ID: 5 Name: Abdul Rahman Invoice No: 11 Amount: 6000
使用 Query.join() 方法可以轻松实现实际的 SQL JOIN 语法，如下所示 –
session.query(Customer).join(Invoice).filter(Invoice.amount == 8500).all()
join 的 SQL 表达式将显示在控制台上 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers JOIN invoices ON customers.id = invoices.custid WHERE invoices.amount = ?
我们可以使用 for 循环遍历结果 –
result = session.query(Customer).join(Invoice).filter(Invoice.amount == 8500)
for row in result:
for inv in row.invoices:
print (row.id, row.name, inv.invno, inv.amount)
使用 8500 作为绑定参数，显示以下输出 –
4 Govind Kala 8 8500
Query.join() 知道如何在这些表之间进行连接，因为它们之间只有一个外键。如果没有外键或更多外键，Query.join() 在使用以下形式之一时效果更好 –
| query.join(Invoice, id == Address.custid) | 显式条件 |
| query.join(Customer.invoices) | 指定从左到右的关系 |
| query.join(Invoice, Customer.invoices) | 相同，有明确的目标 |
| query.join(‘invoices’) | 同样，使用字符串 |
类似地，可以使用 outerjoin() 函数来实现左外连接。
query.outerjoin(Customer.invoices)
subquery() 方法生成一个 SQL 表达式，表示嵌入在别名中的 SELECT 语句。
from sqlalchemy.sql import func
stmt = session.query(
Invoice.custid, func.count('*').label('invoice_count')
).group_by(Invoice.custid).subquery()
stmt 对象将包含一个 SQL 语句，如下所示 –
SELECT invoices.custid, count(:count_1) AS invoice_count FROM invoices GROUP BY invoices.custid
一旦我们有了我们的语句，它的行为就像一个表结构。语句中的列可通过名为 c 的属性访问，如下面的代码所示 –
for u, count in session.query(Customer, stmt.c.invoice_count).outerjoin(stmt, Customer.id == stmt.c.custid).order_by(Customer.id): print(u.name, count)
上面的 for 循环显示发票的名称计数如下 –
Arjun Pandit None Gopal Krishna 2 Govind Pant 2 Govind Kala 2 Abdul Rahman 2
公共关系运算符
在本章中，我们将讨论建立在关系上的运算符。
__eq__()
上面的运算符是一个多对一的“等于”比较。此运算符的代码行如下所示 –
s = session.query(Customer).filter(Invoice.invno.__eq__(12))
上述代码行的等效 SQL 查询是 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers, invoices WHERE invoices.invno = ?
__ne__()
这个运算符是一个多对一的“不等于”比较。此运算符的代码行如下所示 –
s = session.query(Customer).filter(Invoice.custid.__ne__(2))
下面给出了上述代码行的等效 SQL 查询 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers, invoices WHERE invoices.custid != ?
包含（）
此运算符用于一对多集合，下面给出的是 contains() 的代码 –
s = session.query(Invoice).filter(Invoice.invno.contains([3,4,5]))
上述代码行的等效 SQL 查询是 –
SELECT invoices.id AS invoices_id, invoices.custid AS invoices_custid, invoices.invno AS invoices_invno, invoices.amount AS invoices_amount FROM invoices WHERE (invoices.invno LIKE '%' + ? || '%')
任何（）
any() 运算符用于集合，如下所示 –
s = session.query(Customer).filter(Customer.invoices.any(Invoice.invno==11))
上述代码行的等效 SQL 查询如下所示 –
SELECT customers.id AS customers_id, customers.name AS customers_name, customers.address AS customers_address, customers.email AS customers_email FROM customers WHERE EXISTS ( SELECT 1 FROM invoices WHERE customers.id = invoices.custid AND invoices.invno = ?)
具有（）
此运算符用于标量引用，如下所示 –
s = session.query(Invoice).filter(Invoice.customer.has(name = 'Arjun Pandit'))
上述代码行的等效 SQL 查询是 –
SELECT invoices.id AS invoices_id, invoices.custid AS invoices_custid, invoices.invno AS invoices_invno, invoices.amount AS invoices_amount FROM invoices WHERE EXISTS ( SELECT 1 FROM customers WHERE customers.id = invoices.custid AND customers.name = ?)
SQLAlchemy ORM – 预加载
急切负载减少了查询的数量。SQLAlchemy 提供了通过查询选项调用的预加载函数，这些函数为查询提供了额外的指令。这些选项决定了如何通过 Query.options() 方法加载各种属性。
子查询加载
我们希望 Customer.invoices 应该立即加载。orm.subqueryload() 选项提供了第二个 SELECT 语句，该语句完全加载与刚刚加载的结果相关联的集合。名称“子查询”导致 SELECT 语句通过重用的 Query 直接构造，并作为子查询嵌入到针对相关表的 SELECT 中。
from sqlalchemy.orm import subqueryload c1 = session.query(Customer).options(subqueryload(Customer.invoices)).filter_by(name = 'Govind Pant').one()
这导致以下两个 SQL 表达式 –
SELECT customers.id
AS customers_id, customers.name
AS customers_name, customers.address
AS customers_address, customers.email
AS customers_email
FROM customers
WHERE customers.name = ?
('Govind Pant',)
SELECT invoices.id
AS invoices_id, invoices.custid
AS invoices_custid, invoices.invno
AS invoices_invno, invoices.amount
AS invoices_amount, anon_1.customers_id
AS anon_1_customers_id
FROM (
SELECT customers.id
AS customers_id
FROM customers
WHERE customers.name = ?)
AS anon_1
JOIN invoices
ON anon_1.customers_id = invoices.custid
ORDER BY anon_1.customers_id, invoices.id 2018-06-25 18:24:47,479
INFO sqlalchemy.engine.base.Engine ('Govind Pant',)
要访问两个表中的数据，我们可以使用以下程序 –
print (c1.name, c1.address, c1.email)
for x in c1.invoices:
print ("Invoice no : {}, Amount : {}".format(x.invno, x.amount))
上述程序的输出如下 –
Govind Pant Gulmandi Aurangabad [email protected] Invoice no : 3, Amount : 10000 Invoice no : 4, Amount : 5000
连接负载
另一个函数称为 orm.joinedload()。这会发出一个 LEFT OUTER JOIN。一步加载引导对象以及相关对象或集合。
from sqlalchemy.orm import joinedload c1 = session.query(Customer).options(joinedload(Customer.invoices)).filter_by(name='Govind Pant').one()
这发出以下表达式，给出与上述相同的输出 –
SELECT customers.id
AS customers_id, customers.name
AS customers_name, customers.address
AS customers_address, customers.email
AS customers_email, invoices_1.id
AS invoices_1_id, invoices_1.custid
AS invoices_1_custid, invoices_1.invno
AS invoices_1_invno, invoices_1.amount
AS invoices_1_amount
FROM customers
LEFT OUTER JOIN invoices
AS invoices_1
ON customers.id = invoices_1.custid
WHERE customers.name = ? ORDER BY invoices_1.id
('Govind Pant',)
OUTER JOIN 产生了两行，但它返回了一个 Customer 实例。这是因为 Query 基于对象标识对返回的实体应用“唯一”策略。可以在不影响查询结果的情况下应用加入的预先加载。
subqueryload() 更适合加载相关集合，而joinedload() 更适合多对一关系。
SQLAlchemy ORM – 删除相关对象
对单个表进行删除操作很容易。您所要做的就是从会话中删除映射类的对象并提交操作。但是，对多个相关表的删除操作有点棘手。
在我们的 sales.db 数据库中，Customer 和 Invoice 类通过一对多类型的关系映射到 customer 和 invoice 表。我们将尝试删除 Customer 对象并查看结果。
作为快速参考，以下是 Customer 和 Invoice 类的定义 –
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///sales.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key = True)
name = Column(String)
address = Column(String)
email = Column(String)
class Invoice(Base):
__tablename__ = 'invoices'
id = Column(Integer, primary_key = True)
custid = Column(Integer, ForeignKey('customers.id'))
invno = Column(Integer)
amount = Column(Integer)
customer = relationship("Customer", back_populates = "invoices")
Customer.invoices = relationship("Invoice", order_by = Invoice.id, back_populates = "customer")
我们设置一个会话并通过使用以下程序使用主 ID 查询它来获取一个 Customer 对象 –
from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind=engine) session = Session() x = session.query(Customer).get(2)
在我们的示例表中，x.name 恰好是“Gopal Krishna”。让我们从会话中删除这个 x 并计算这个名称的出现次数。
session.delete(x) session.query(Customer).filter_by(name = 'Gopal Krishna').count()
生成的 SQL 表达式将返回 0。
SELECT count(*)
AS count_1
FROM (
SELECT customers.id
AS customers_id, customers.name
AS customers_name, customers.address
AS customers_address, customers.email
AS customers_email
FROM customers
WHERE customers.name = ?)
AS anon_1('Gopal Krishna',) 0
但是，x 的相关 Invoice 对象仍然存在。可以通过以下代码验证 –
session.query(Invoice).filter(Invoice.invno.in_([10,14])).count()
此处，10 和 14 是属于客户 Gopal Krishna 的发票编号。上面查询的结果是2，表示相关对象没有被删除。
SELECT count(*) AS count_1 FROM ( SELECT invoices.id AS invoices_id, invoices.custid AS invoices_custid, invoices.invno AS invoices_invno, invoices.amount AS invoices_amount FROM invoices WHERE invoices.invno IN (?, ?)) AS anon_1(10, 14) 2
这是因为 SQLAlchemy 不假设删除级联；我们必须给出一个命令来删除它。
为了改变行为，我们在 User.addresses 关系上配置级联选项。让我们关闭正在进行的会话，使用新的 declarative_base() 并重新声明 User 类，添加包括级联配置在内的地址关系。
关系函数中的级联属性是一个以逗号分隔的级联规则列表，它决定了会话操作应该如何从父级到子级“级联”。默认为False，表示是“save-update,merge”。
可用的级联如下 –
- 保存更新
- 合并
- 清除
- 删除
- 删除孤儿
- 刷新-过期
经常使用的选项是“all, delete-orphan”，表示相关对象在所有情况下都应该跟随父对象，并在解除关联时删除。
因此，重新声明的 Customer 类如下所示 –
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key = True)
name = Column(String)
address = Column(String)
email = Column(String)
invoices = relationship(
"Invoice",
order_by = Invoice.id,
back_populates = "customer",
cascade = "all,
delete, delete-orphan"
)
让我们使用以下程序删除带有 Gopal Krishna 名称的 Customer，并查看其相关 Invoice 对象的数量 –
from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind = engine) session = Session() x = session.query(Customer).get(2) session.delete(x) session.query(Customer).filter_by(name = 'Gopal Krishna').count() session.query(Invoice).filter(Invoice.invno.in_([10,14])).count()
计数现在为 0，上面的脚本发出以下 SQL –
SELECT customers.id
AS customers_id, customers.name
AS customers_name, customers.address
AS customers_address, customers.email
AS customers_email
FROM customers
WHERE customers.id = ?
(2,)
SELECT invoices.id
AS invoices_id, invoices.custid
AS invoices_custid, invoices.invno
AS invoices_invno, invoices.amount
AS invoices_amount
FROM invoices
WHERE ? = invoices.custid
ORDER BY invoices.id (2,)
DELETE FROM invoices
WHERE invoices.id = ? ((1,), (2,))
DELETE FROM customers
WHERE customers.id = ? (2,)
SELECT count(*)
AS count_1
FROM (
SELECT customers.id
AS customers_id, customers.name
AS customers_name, customers.address
AS customers_address, customers.email
AS customers_email
FROM customers
WHERE customers.name = ?)
AS anon_1('Gopal Krishna',)
SELECT count(*)
AS count_1
FROM (
SELECT invoices.id
AS invoices_id, invoices.custid
AS invoices_custid, invoices.invno
AS invoices_invno, invoices.amount
AS invoices_amount
FROM invoices
WHERE invoices.invno IN (?, ?))
AS anon_1(10, 14)
0
多对多关系
两个表之间的多对多关系是通过添加一个关联表来实现的，它有两个外键——一个来自每个表的主键。此外，映射到这两个表的类具有一个属性，该属性具有其他关联表的对象集合，作为relationship() 函数的次要属性。
为此，我们将创建一个 SQLite 数据库 (mycollege.db)，其中包含两个表 – 部门和员工。在这里，我们假设一个员工属于多个部门，一个部门有多个员工。这构成了多对多的关系。
映射到部门和员工表的员工和部门类的定义如下 –
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
engine = create_engine('sqlite:///mycollege.db', echo = True)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy.orm import relationship
class Department(Base):
__tablename__ = 'department'
id = Column(Integer, primary_key = True)
name = Column(String)
employees = relationship('Employee', secondary = 'link')
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key = True)
name = Column(String)
departments = relationship(Department,secondary='link')
我们现在定义一个 Link 类。它链接到链接表并包含分别引用部门和员工表的主键的department_id 和employee_id 属性。
class Link(Base):
__tablename__ = 'link'
department_id = Column(
Integer,
ForeignKey('department.id'),
primary_key = True)
employee_id = Column(
Integer,
ForeignKey('employee.id'),
primary_key = True)
在这里，我们必须注意，Department 类具有与 Employee 类相关的雇员属性。关系函数的次要属性被分配一个链接作为其值。
类似地，Employee 类具有与Department 类相关的departments 属性。关系函数的次要属性被分配一个链接作为其值。
所有这三个表都是在执行以下语句时创建的 –
Base.metadata.create_all(engine)
Python 控制台发出以下 CREATE TABLE 查询 –
CREATE TABLE department ( id INTEGER NOT NULL, name VARCHAR, PRIMARY KEY (id) ) CREATE TABLE employee ( id INTEGER NOT NULL, name VARCHAR, PRIMARY KEY (id) ) CREATE TABLE link ( department_id INTEGER NOT NULL, employee_id INTEGER NOT NULL, PRIMARY KEY (department_id, employee_id), FOREIGN KEY(department_id) REFERENCES department (id), FOREIGN KEY(employee_id) REFERENCES employee (id) )
我们可以通过使用 SQLiteStudio 打开 mycollege.db 来检查这一点，如下面的屏幕截图所示 –
接下来我们创建 Department 类的三个对象和 Employee 类的三个对象，如下所示 –
d1 = Department(name = "Accounts") d2 = Department(name = "Sales") d3 = Department(name = "Marketing") e1 = Employee(name = "John") e2 = Employee(name = "Tony") e3 = Employee(name = "Graham")
每个表都有一个带有 append() 方法的集合属性。我们可以将 Employee 对象添加到 Department 对象的 Employees 集合中。同样，我们可以将 Department 对象添加到 Employee 对象的部门集合属性中。
e1.departments.append(d1) e2.departments.append(d3) d1.employees.append(e3) d2.employees.append(e2) d3.employees.append(e1) e3.departments.append(d2)
我们现在要做的就是设置一个会话对象，向其中添加所有对象并提交更改，如下所示 –
from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind = engine) session = Session() session.add(e1) session.add(e2) session.add(d1) session.add(d2) session.add(d3) session.add(e3) session.commit()
以下 SQL 语句将在 Python 控制台上发出 –
INSERT INTO department (name) VALUES (?) ('Accounts',)
INSERT INTO department (name) VALUES (?) ('Sales',)
INSERT INTO department (name) VALUES (?) ('Marketing',)
INSERT INTO employee (name) VALUES (?) ('John',)
INSERT INTO employee (name) VALUES (?) ('Graham',)
INSERT INTO employee (name) VALUES (?) ('Tony',)
INSERT INTO link (department_id, employee_id) VALUES (?, ?) ((1, 2), (3, 1), (2, 3))
INSERT INTO link (department_id, employee_id) VALUES (?, ?) ((1, 1), (2, 2), (3, 3))
要检查上述操作的效果，请使用 SQLiteStudio 并查看部门、员工和链接表中的数据 –
要显示数据，请运行以下查询语句 –
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = engine)
session = Session()
for x in session.query( Department, Employee).filter(Link.department_id == Department.id,
Link.employee_id == Employee.id).order_by(Link.department_id).all():
print ("Department: {} Name: {}".format(x.Department.name, x.Employee.name))
根据我们示例中填充的数据，输出将显示如下 –
Department: Accounts Name: John Department: Accounts Name: Graham Department: Sales Name: Graham Department: Sales Name: Tony Department: Marketing Name: John Department: Marketing Name: Tony
SQLAlchemy – 方言
SQLAlchemy 使用方言系统与各种类型的数据库进行通信。每个数据库都有一个对应的 DBAPI 包装器。所有方言都要求安装适当的 DBAPI 驱动程序。
SQLAlchemy API 中包含以下方言 –
- 火鸟
- 微软 SQL 服务器
- MySQL
- 甲骨文
- PostgreSQL
- SQL
- 赛贝斯
基于 URL 的 Engine 对象由 create_engine() 函数生成。这些 URL 可以包括用户名、密码、主机名和数据库名称。可能有可选的关键字参数用于其他配置。在某些情况下，接受文件路径，而在其他情况下，“数据源名称”替换“主机”和“数据库”部分。数据库 URL 的典型形式如下 –
dialect+driver://username:password@host:port/database
PostgreSQL
PostgreSQL 方言使用psycopg2作为默认 DBAPI。pg8000 也可用作纯 Python 替代品，如下所示：
# default
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase')
# psycopg2
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')
# pg8000
engine = create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')
MySQL
MySQL 方言使用mysql-python作为默认 DBAPI。有许多 MySQL DBAPI 可用，例如 MySQL-connector-python，如下所示 –
# default
engine = create_engine('mysql://scott:tiger@localhost/foo')
# mysql-python
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')
# MySQL-connector-python
engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')
甲骨文
Oracle 方言使用cx_oracle作为默认 DBAPI，如下所示 –
engine = create_engine('oracle://scott:[email protected]:1521/sidname')
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
微软 SQL 服务器
SQL Server 方言使用pyodbc作为默认 DBAPI。pymssql 也可用。
# pyodbc
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn')
# pymssql
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')
SQLite
SQLite默认使用 Python 内置模块sqlite3连接到基于文件的数据库。当 SQLite 连接到本地文件时，URL 格式略有不同。URL 的“文件”部分是数据库的文件名。对于相对文件路径，这需要三个斜杠，如下所示 –
engine = create_engine('sqlite:///foo.db')
对于绝对文件路径，三个斜杠后跟绝对路径，如下所示 –
engine = create_engine('sqlite:///C:\\path\\to\\foo.db')
要使用 SQLite:memory:database，请指定一个空 URL，如下所示 –
engine = create_engine('sqlite://')
结论
在本教程的第一部分中，我们学习了如何使用表达式语言来执行 SQL 语句。表达式语言在 Python 代码中嵌入了 SQL 结构。在第二部分，我们讨论了 SQLAlchemy 的对象关系映射能力。ORM API 使用 Python 类映射 SQL 表。
