SQLAlchemy对象关系映射器提供了一种将用户定义的Python类与数据库表以及这些类(对象)的实例与相应表中的行关联起来的方法。它包括一个透明地同步对象及其相关行之间状态的所有更改的系统,称为 unit of work 以及用用户定义的类及其相互之间定义的关系表示数据库查询的系统。
ORM与用于构造ORM的SQLAlchemy表达式语言形成对比。而SQL表达式语言 SQL表达式语言教程 提出了一种直接表示关系数据库原始结构而不加意见的系统,ORM提出了一种高级抽象的使用模式,它本身就是表达式语言应用的一个例子。
尽管ORM和表达语言的使用模式有重叠,但相似性比最初可能出现的更为肤浅。一种方法是从用户定义的角度来处理数据的结构和内容。 domain model 它透明地持久化并从其底层存储模型中刷新。另一种方法是从文本模式和SQL表达式表示的角度进行处理,这些表达式显式地组合成数据库单独使用的消息。
成功的应用程序可以只用对象关系映射器来构造。在高级情况下,使用ORM构建的应用程序可能会在某些需要特定数据库交互的区域中偶尔直接使用表达式语言。
以下教程采用doctest格式,即 >>>
行表示可以在python命令提示下键入的内容,下面的文本表示预期的返回值。
对于本教程,我们将使用一个只在内存中的sqlite数据库。连接我们使用的 create_engine()
::
>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///:memory:', echo=True)
这个 echo
标志是设置sqlachemy日志的快捷方式,它是通过python的标准实现的。 logging
模块。启用它后,我们将看到生成的所有SQL。如果您正在学习本教程并希望生成的输出更少,请将其设置为 False
. 本教程将在弹出窗口后面格式化SQL,这样它就不会妨碍我们的工作;只需单击“SQL”链接查看正在生成的内容。
的返回值 create_engine()
是的实例 Engine
它表示到数据库的核心接口,通过 dialect 处理数据库的详细信息和 DBAPI 在使用中。在这种情况下,sqlite方言将解释python内置的指令 sqlite3
模块。
第一次像这样的方法 Engine.execute()
或 Engine.connect()
被称为 Engine
建立一个真实的 DBAPI 连接到数据库,该数据库随后用于发出SQL。使用ORM时,我们通常不使用 Engine
直接创建;相反,它被ORM在幕后使用,稍后我们将看到。
参见
数据库URL -包括以下示例 create_engine()
通过指向更多信息的链接连接到几种数据库。
使用ORM时,配置过程首先描述将要处理的数据库表,然后定义将映射到这些表的自己的类。在现代的SQLAlchemy中,这两个任务通常一起执行,使用的系统称为 声明的 ,这允许我们创建包含指令的类来描述它们将映射到的实际数据库表。
使用声明性系统映射的类是根据一个基类定义的,该基类维护一个与该基相关的类和表的目录-这称为 声明性基类 . 我们的应用程序在一个通常导入的模块中通常只有这个基的一个实例。我们使用 declarative_base()
功能如下:
>>> from sqlalchemy.ext.declarative import declarative_base
>>> Base = declarative_base()
既然我们有了一个“基”,就可以根据它定义任意数量的映射类。我们从一张叫 users
,它将使用我们的应用程序为最终用户存储记录。一个名为 User
将是我们映射此表的类。在类中,我们定义了要映射到的表的详细信息,主要是表名以及列的名称和数据类型:
>>> from sqlalchemy import Column, Integer, String
>>> class User(Base):
... __tablename__ = 'users'
...
... id = Column(Integer, primary_key=True)
... name = Column(String)
... fullname = Column(String)
... nickname = Column(String)
...
... def __repr__(self):
... return "<User(name='%s', fullname='%s', nickname='%s')>" % (
... self.name, self.fullname, self.nickname)
至少使用声明性的类需要 __tablename__
属性,以及至少一个 Column
它是主键的一部分 1. SQLAlchemy从不单独对类引用的表进行任何假设,包括它没有名称、数据类型或约束的内置约定。但这并不意味着需要样板文件;相反,我们鼓励您使用助手函数和mixin类创建自己的自动约定,具体描述见 混合和自定义基类 .
当我们的类被构造时,声明性替换了 Column
具有特殊python访问器的对象称为 descriptors ;这是一个被称为 instrumentation . “instructed”映射类将为我们提供在SQL上下文中引用表以及从数据库中持久化和加载列值的方法。
除了映射过程对我们的类所做的工作之外,该类仍然是一个普通的python类,我们可以定义应用程序所需的任意数量的普通属性和方法。
有关为什么需要主键的信息,请参阅 如何映射没有主键的表? .
和我们一起 User
通过声明性系统构造的类,我们已经定义了关于表的信息,称为 table metadata . sqlAlchemy用于表示特定表的此信息的对象称为 Table
对象,这里声明性为我们做了一个。我们可以通过检查 __table__
属性:
>>> User.__table__
Table('users', MetaData(bind=None),
Column('id', Integer(), table=<users>, primary_key=True, nullable=False),
Column('name', String(), table=<users>),
Column('fullname', String(), table=<users>),
Column('nickname', String(), table=<users>), schema=None)
当我们声明类时,声明性使用了一个python元类,以便在类声明完成后执行其他活动;在这个阶段中,它随后创建了一个 Table
对象,并通过构造 Mapper
对象。这个对象是一个我们通常不需要直接处理的幕后对象(尽管它可以在我们需要的时候提供关于我们的映射的大量信息)。
这个 Table
对象是称为 MetaData
. 使用声明性时,可以使用 .metadata
声明性基类的属性。
这个 MetaData
是一个 registry 其中包括向数据库发出一组有限的模式生成命令的能力。因为我们的sqlite数据库实际上没有 users
有表格,我们可以用 MetaData
为尚未存在的所有表向数据库发出create table语句。下面,我们称之为 MetaData.create_all()
方法,传入 Engine
作为数据库连接的来源。我们将看到首先发出特殊命令来检查 users
表,然后是 CREATE TABLE
声明:
>>> Base.metadata.create_all(engine)
SELECT ...
PRAGMA table_info("users")
()
CREATE TABLE users (
id INTEGER NOT NULL, name VARCHAR,
fullname VARCHAR,
nickname VARCHAR,
PRIMARY KEY (id)
)
()
COMMIT
最小表描述与完整描述
熟悉create table语法的用户可能会注意到varchar列的生成没有长度;在sqlite和postgresql上,这是一个有效的数据类型,但在其他数据类型上,这是不允许的。因此,如果在其中一个数据库上运行本教程,并且您希望使用sqlachemy来发布create table,那么可以为 String
类型如下:
Column(String(50))
上的长度字段 String
以及类似的精度/比例字段 Integer
, Numeric
除创建表时外,SQLAlchemy不引用等。
此外,Firebird和Oracle需要序列来生成新的主键标识符,而sqlAlchemy在没有得到指示的情况下不会生成或假定这些标识符。为此,您使用 Sequence
结构:
from sqlalchemy import Sequence
Column(Integer, Sequence('user_id_seq'), primary_key=True)
一个完整的,万无一失的 Table
因此,通过声明性映射生成的是:
class User(Base):
__tablename__ = 'users'
id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
name = Column(String(50))
fullname = Column(String(50))
nickname = Column(String(50))
def __repr__(self):
return "<User(name='%s', fullname='%s', nickname='%s')>" % (
self.name, self.fullname, self.nickname)
我们分别包含这个更详细的表定义,以突出主要面向Python使用的最小构造与将用于在具有更严格要求的特定后端集上发出create table语句的构造之间的区别。
映射完成后,现在让我们创建并检查 User
对象:
>>> ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
>>> ed_user.name
'ed'
>>> ed_user.nickname
'edsnickname'
>>> str(ed_user.id)
'None'
即使我们没有在构造函数中指定它, id
属性仍然产生值 None
当我们访问它时(与Python通常的提升行为相反) AttributeError
对于未定义的属性)。SQLAlchemy instrumentation 通常在第一次访问列映射属性时生成此默认值。对于那些我们已经实际分配了值的属性,检测系统正在跟踪这些分配,以便在最终要发送到数据库的insert语句中使用。
我们现在准备开始与数据库对话。ORM对数据库的“句柄”是 Session
. 当我们第一次设置应用程序时,与我们的 create_engine()
语句,我们定义 Session
将用作新工厂的类 Session
对象::
>>> from sqlalchemy.orm import sessionmaker
>>> Session = sessionmaker(bind=engine)
如果您的应用程序还没有 Engine
定义模块级对象时,只需如下设置:
>>> Session = sessionmaker()
稍后,当您使用 create_engine()
,连接到 Session
使用 configure()
::
>>> Session.configure(bind=engine) # once engine is available
这个定制的 Session
类将创建新的 Session
绑定到数据库的对象。调用时可以定义其他事务性特征。 sessionmaker
同样;这些在后面的章节中描述。然后,每当需要与数据库进行对话时,都要实例化一个 Session
::
>>> session = Session()
以上 Session
与启用的sqlite关联 Engine
,但尚未打开任何连接。第一次使用时,它从由 Engine
,并一直保留到我们提交所有更改和/或关闭会话对象。
坚持我们的 User
对象,我们 add()
它给我们 Session
::
>>> ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
>>> session.add(ed_user)
此时,我们说实例是 悬而未决的 ;尚未发出SQL,对象尚未由数据库中的一行表示。这个 Session
将发出SQL以保持 Ed Jones
一旦需要,使用一个称为 脸红 . 如果我们查询数据库 Ed Jones
,将首先刷新所有挂起的信息,然后立即发出查询。
例如,下面我们创建一个新的 Query
加载的实例的对象 User
. 我们“过滤”的 name
属性 ed
,并指示我们只希望在完整的行列表中得到第一个结果。一 User
返回的实例与我们添加的实例相同:
sql>>> our_user = session.query(User).filter_by(name='ed').first()
BEGIN (implicit)
INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
('ed', 'Ed Jones', 'edsnickname')
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ?
LIMIT ? OFFSET ?
('ed', 1, 0)
>>> our_user
<User(name='ed', fullname='Ed Jones', nickname='edsnickname')>
事实上, Session
已确定返回的行是 same 行作为一个已经在其对象的内部映射中表示的实例,因此我们实际返回了与刚才添加的实例相同的实例::
>>> ed_user is our_user
True
这里的ORM概念被称为 identity map 并确保在 Session
对同一组数据进行操作。一旦具有特定主键的对象出现在 Session
,所有SQL查询 Session
将始终为该特定的主键返回同一个python对象;如果试图在会话中放置第二个已持久化且具有相同主键的对象,也会引发错误。
我们可以增加更多 User
对象同时使用 add_all()
:
>>> session.add_all([
... User(name='wendy', fullname='Wendy Williams', nickname='windy'),
... User(name='mary', fullname='Mary Contrary', nickname='mary'),
... User(name='fred', fullname='Fred Flintstone', nickname='freddy')])
另外,我们已经决定了Ed的昵称不是很好,所以我们来改变它:
>>> ed_user.nickname = 'eddie'
这个 Session
正在关注。例如,它知道 Ed Jones
已修改:
>>> session.dirty
IdentitySet([<User(name='ed', fullname='Ed Jones', nickname='eddie')>])
那三个新的 User
对象挂起:
>>> session.new
IdentitySet([<User(name='wendy', fullname='Wendy Williams', nickname='windy')>,
<User(name='mary', fullname='Mary Contrary', nickname='mary')>,
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>])
我们告诉 Session
我们希望发布对数据库的所有剩余更改,并提交整个过程中一直在进行的事务。我们这样做通过 commit()
. 这个 Session
发出 UPDATE
“ed”上昵称更改的声明,以及 INSERT
三个新的声明 User
我们添加的对象:
sql>>> session.commit()
UPDATE users SET nickname=? WHERE users.id = ?
('eddie', 1)
INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
('wendy', 'Wendy Williams', 'windy')
INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
('mary', 'Mary Contrary', 'mary')
INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
('fred', 'Fred Flintstone', 'freddy')
COMMIT
commit()
刷新对数据库的其余更改,并提交事务。会话引用的连接资源现在返回到连接池。此会话的后续操作将在 new 事务,它将在第一次需要时重新获取连接资源。
如果我们看看Ed的 id
属性,之前是 None
,它现在有一个值:
sql>>> ed_user.id
BEGIN (implicit)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.id = ?
(1,)
1
后 Session
在数据库中插入新行,所有新生成的标识符和数据库生成的默认值都将在实例上立即可用,或者在第一次访问时加载。在这种情况下,整个行在Access上被重新加载,因为在我们发出之后开始了一个新事务。 commit()
. 默认情况下,当第一次在新事务中访问前一个事务时,sqlAlchemy将刷新该事务中的数据,以使最新状态可用。重新加载的级别是可配置的,如中所述。 使用会话 .
会话对象状态
作为我们 User
从外部移动的对象 Session
到里面 Session
如果没有主键,实际插入时,它将在四个可用“对象状态”中的三个之间移动- 瞬态 , 悬而未决的 和 持久的 . 了解这些状态及其含义始终是一个好主意-一定要阅读 Quickie对象状态简介 快速概述。
自从 Session
在事务中工作,我们也可以回滚所做的更改。让我们做两个我们将要恢复的更改; ed_user
的用户名设置为 Edwardo
:
>>> ed_user.name = 'Edwardo'
我们将添加另一个错误的用户, fake_user
:
>>> fake_user = User(name='fakeuser', fullname='Invalid', nickname='12345')
>>> session.add(fake_user)
查询会话时,可以看到它们被刷新到当前事务中:
sql>>> session.query(User).filter(User.name.in_(['Edwardo', 'fakeuser'])).all()
UPDATE users SET name=? WHERE users.id = ?
('Edwardo', 1)
INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
('fakeuser', 'Invalid', '12345')
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name IN (?, ?)
('Edwardo', 'fakeuser')
[<User(name='Edwardo', fullname='Ed Jones', nickname='eddie')>, <User(name='fakeuser', fullname='Invalid', nickname='12345')>]
回过头来,我们可以看到 ed_user
的名称返回到 ed
和 fake_user
已退出会话:
发出选择说明对数据库所做的更改:
sql>>> session.query(User).filter(User.name.in_(['ed', 'fakeuser'])).all()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name IN (?, ?)
('ed', 'fakeuser')
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
A Query
对象是使用 query()
方法对 Session
. 此函数接受可变数量的参数,这些参数可以是类和类插入描述符的任意组合。下面,我们指出 Query
哪些负载 User
实例。当在迭代上下文中进行评估时, User
返回存在的对象:
sql>>> for instance in session.query(User).order_by(User.id):
... print(instance.name, instance.fullname)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users ORDER BY users.id
()
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flintstone
这个 Query
还接受ORM插入描述符作为参数。当多个类实体或基于列的实体表示为 query()
函数,返回结果表示为元组:
sql>>> for name, fullname in session.query(User.name, User.fullname):
... print(name, fullname)
SELECT users.name AS users_name,
users.fullname AS users_fullname
FROM users
()
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flintstone
返回的元组 Query
是 已命名 元组,由 KeyedTuple
类,并且可以像普通的Python对象一样进行处理。这些名称与属性的名称和类的类名相同:
sql>>> for row in session.query(User, User.name).all():
... print(row.User, row.name)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
()
<User(name='ed', fullname='Ed Jones', nickname='eddie')> ed
<User(name='wendy', fullname='Wendy Williams', nickname='windy')> wendy
<User(name='mary', fullname='Mary Contrary', nickname='mary')> mary
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')> fred
可以使用 label()
构造,可从任何 ColumnElement
-派生对象以及映射到一个对象的任何类属性(例如 User.name
):
sql>>> for row in session.query(User.name.label('name_label')).all():
... print(row.name_label)
SELECT users.name AS name_label
FROM users
()ed
wendy
mary
fred
给予完整实体的名称,如 User
,假设调用中存在多个实体 query()
,可以使用 aliased()
:
>>> from sqlalchemy.orm import aliased
>>> user_alias = aliased(User, name='user_alias')
sql>>> for row in session.query(user_alias, user_alias.name).all():
... print(row.user_alias)
SELECT user_alias.id AS user_alias_id,
user_alias.name AS user_alias_name,
user_alias.fullname AS user_alias_fullname,
user_alias.nickname AS user_alias_nickname
FROM users AS user_alias
()<User(name='ed', fullname='Ed Jones', nickname='eddie')>
<User(name='wendy', fullname='Wendy Williams', nickname='windy')>
<User(name='mary', fullname='Mary Contrary', nickname='mary')>
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>
基本操作 Query
包括发布限制和偏移,最方便地使用python数组切片,通常与order by结合使用:
sql>>> for u in session.query(User).order_by(User.id)[1:3]:
... print(u)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users ORDER BY users.id
LIMIT ? OFFSET ?
(2, 1)<User(name='wendy', fullname='Wendy Williams', nickname='windy')>
<User(name='mary', fullname='Mary Contrary', nickname='mary')>
以及过滤结果 filter_by()
,使用关键字参数:
sql>>> for name, in session.query(User.name).\
... filter_by(fullname='Ed Jones'):
... print(name)
SELECT users.name AS users_name FROM users
WHERE users.fullname = ?
('Ed Jones',)
ed
或… filter()
,它使用更灵活的SQL表达式语言构造。这些允许您在映射类上使用具有类级属性的常规python运算符:
sql>>> for name, in session.query(User.name).\
... filter(User.fullname=='Ed Jones'):
... print(name)
SELECT users.name AS users_name FROM users
WHERE users.fullname = ?
('Ed Jones',)
ed
这个 Query
对象完全 生成的 ,这意味着大多数方法调用都返回一个新的 Query
对象,在此对象上可以添加其他条件。例如,要查询名为“ed”且全名为“ed jones”的用户,可以调用 filter()
两次,使用 AND
:
sql>>> for user in session.query(User).\
... filter(User.name=='ed').\
... filter(User.fullname=='Ed Jones'):
... print(user)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ? AND users.fullname = ?
('ed', 'Ed Jones')
<User(name='ed', fullname='Ed Jones', nickname='eddie')>
下面是在 filter()
:
equals
::
query.filter(User.name == 'ed')
not equals
::
query.filter(User.name != 'ed')
LIKE
::
query.filter(User.name.like('%ed%'))
注解
ColumnOperators.like()
呈现like运算符,它在某些后端不区分大小写,在其他后端区分大小写。对于保证不区分大小写的比较,请使用ColumnOperators.ilike()
.
ILIKE
(不区分大小写,例如)::
query.filter(User.name.ilike('%ed%'))
注解
大多数后端不直接支持iLike。对于那些
ColumnOperators.ilike()
运算符呈现一个表达式,该表达式与应用于每个操作数的下SQL函数组合在一起。
IN
::
query.filter(User.name.in_(['ed', 'wendy', 'jack']))
# works with query objects too:
query.filter(User.name.in_(
session.query(User.name).filter(User.name.like('%ed%'))
))
NOT IN
::
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
IS NULL
::
query.filter(User.name == None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))
IS NOT NULL
::
query.filter(User.name != None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))
AND
::
# use and_()
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
# or chain multiple filter()/filter_by() calls
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
注解
确保使用
and_()
和 not Pythonand
接线员!
OR
::
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))
注解
确保使用
or_()
和 not Pythonor
接线员!
MATCH
::
query.filter(User.name.match('wendy'))
注解
match()
使用特定于数据库的MATCH
或CONTAINS
函数;其行为因后端而异,在某些后端(如sqlite)上不可用。
许多方法 Query
立即发出SQL并返回包含已加载数据库结果的值。下面是一个简短的旅行:
all()
返回一个列表:
>>> query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
sql>>> query.all()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name LIKE ? ORDER BY users.id
('%ed',)
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>,
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>]
first()
应用一个限制并以标量形式返回第一个结果:
sql>>> query.first()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name LIKE ? ORDER BY users.id
LIMIT ? OFFSET ?
('%ed', 1, 0)
<User(name='ed', fullname='Ed Jones', nickname='eddie')>
one()
完全获取所有行,如果结果中不存在一个对象标识或复合行,则会引发错误。找到多行时:
>>> user = query.one()
Traceback (most recent call last):
...
MultipleResultsFound: Multiple rows were found for one()
找不到行:
>>> user = query.filter(User.id == 99).one()
Traceback (most recent call last):
...
NoResultFound: No row was found for one()
这个 one()
对于希望处理“未找到项目”和“找到多个项目”不同的系统来说,方法是很好的;例如RESTful Web服务,它可能希望在未找到结果时引发“404未找到”,但在找到多个结果时引发应用程序错误。
one_or_none()
就像 one()
,但如果没有找到结果,则不会引发错误;它只是返回 None
. 喜欢 one()
但是,如果发现多个结果,则会引发错误。
scalar()
调用 one()
方法,并在成功时返回行的第一列:
>>> query = session.query(User.id).filter(User.name == 'ed').\
... order_by(User.id)
sql>>> query.scalar()
SELECT users.id AS users_id
FROM users
WHERE users.name = ? ORDER BY users.id
('ed',)
1
文字字符串可以灵活地用于 Query
,通过指定它们与 text()
构造,这是最适用的方法所接受的。例如, filter()
和 order_by()
:
>>> from sqlalchemy import text
sql>>> for user in session.query(User).\
... filter(text("id<224")).\
... order_by(text("id")).all():
... print(user.name)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE id<224 ORDER BY id
()
ed
wendy
mary
fred
可以使用基于字符串的SQL,使用冒号指定绑定参数。要指定值,请使用 params()
方法:
sql>>> session.query(User).filter(text("id<:value and name=:name")).\
... params(value=224, name='fred').order_by(User.id).one()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE id<? and name=? ORDER BY users.id
(224, 'fred')
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')>
要使用完全基于字符串的语句,请 text()
无法将表示完整语句的构造传递给 from_statement()
. 如果没有其他说明符,则字符串SQL中的列将与基于名称的模型列匹配,例如下面我们仅使用星号表示加载所有列:
sql>>> session.query(User).from_statement(
... text("SELECT * FROM users where name=:name")).\
... params(name='ed').all()
SELECT * FROM users where name=?
('ed',)
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
名称上的匹配列适用于简单情况,但在处理包含重复列名的复杂语句时,或者在使用不容易与特定名称匹配的匿名ORM构造时,可能会变得不方便。此外,在我们的映射列中存在类型行为,我们在处理结果行时可能会发现这是必需的。对于这些情况, text()
构造允许我们将其文本SQL按位置链接到核心或ORM映射的列表达式;我们可以通过将列表达式作为位置参数传递给 TextClause.columns()
方法:
>>> stmt = text("SELECT name, id, fullname, nickname "
... "FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id, User.fullname, User.nickname)
sql>>> session.query(User).from_statement(stmt).params(name='ed').all()
SELECT name, id, fullname, nickname FROM users where name=?
('ed',)
[<User(name='ed', fullname='Ed Jones', nickname='eddie')>]
1.1 新版功能: 这个 TextClause.columns()
方法现在接受将按位置与纯文本SQL结果集匹配的列表达式,从而消除了对列名称进行匹配的需要,甚至在SQL语句中是唯一的。
从中选择时 text()
构建 Query
仍可以指定要返回的列和实体;而不是 query(User)
我们也可以单独要求列,在任何其他情况下:
>>> stmt = text("SELECT name, id FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id)
sql>>> session.query(User.id, User.name).\
... from_statement(stmt).params(name='ed').all()
SELECT name, id FROM users where name=?
('ed',)
[(1, u'ed')]
sql>>> session.query(User).filter(User.name.like('%ed')).count()
SELECT count(*) AS count_1
FROM (SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name LIKE ?) AS anon_1
('%ed',)
2
这个 count()
方法用于确定SQL语句将返回多少行。查看上面生成的SQL,sqlAlchemy总是将我们正在查询的内容放入子查询中,然后计算其中的行数。在某些情况下,这可以简化为 SELECT count(*) FROM table
但是,现代版本的SQLAlchemy并不试图猜测这是什么时候合适,因为可以使用更明确的方法发出准确的sql。
对于需要特别指出“要计数的东西”的情况,我们可以直接使用表达式指定“count”函数。 func.count()
,可从 func
构建。下面我们使用它返回每个不同用户名的计数:
>>> from sqlalchemy import func
sql>>> session.query(func.count(User.name), User.name).group_by(User.name).all()
SELECT count(users.name) AS count_1, users.name AS users_name
FROM users GROUP BY users.name
()
[(1, u'ed'), (1, u'fred'), (1, u'mary'), (1, u'wendy')]
为了实现我们的简单 SELECT count(*) FROM table
我们可以将其应用于:
sql>>> session.query(func.count('*')).select_from(User).scalar()
SELECT count(?) AS count_1
FROM users
('*',)
4
用法 select_from()
如果我们用 User
主键直接:
sql>>> session.query(func.count(User.id)).scalar()
SELECT count(users.id) AS count_1
FROM users
()
4
让我们考虑第二个表如何与 User
,可以进行映射和查询。我们系统中的用户可以存储与其用户名关联的任意数量的电子邮件地址。这意味着从 users
到存储电子邮件地址的新表,我们将调用该表 addresses
. 使用声明性,我们定义这个表及其映射类, Address
:
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy.orm import relationship
>>> class Address(Base):
... __tablename__ = 'addresses'
... id = Column(Integer, primary_key=True)
... email_address = Column(String, nullable=False)
... user_id = Column(Integer, ForeignKey('users.id'))
...
... user = relationship("User", back_populates="addresses")
...
... def __repr__(self):
... return "<Address(email_address='%s')>" % self.email_address
>>> User.addresses = relationship(
... "Address", order_by=Address.id, back_populates="user")
上面的课程介绍了 ForeignKey
构造,这是应用于 Column
指示此列中的值应为 constrained 是命名远程列中存在的值。这是关系数据库的一个核心特性,也是将原本不相连的表集合转换为具有丰富重叠关系的“粘合剂”。这个 ForeignKey
上面表示的值 addresses.user_id
列应约束为 users.id
列,即其主键。
第二个指令,称为 relationship()
,告诉ORM Address
类本身应链接到 User
类,使用属性 Address.user
. relationship()
使用两个表之间的外键关系来确定此链接的性质,从而确定 Address.user
将 many to one . 额外的 relationship()
指令放在 User
属性下的映射类 User.addresses
. 在两者 relationship()
指令,参数 relationship.back_populates
被分配来引用互补的属性名;通过这样做,每个属性名 relationship()
能够做出明智的决定,与逆向表达的关系相同;一方面, Address.user
指的是 User
另一方面, User.addresses
指的是 Address
实例。
注解
这个 relationship.back_populates
参数是名为 relationship.backref
. 这个 relationship.backref
参数没有移动到任何地方,将始终保持可用!这个 relationship.back_populates
是一样的,只是有点冗长,更容易操作。有关整个主题的概述,请参阅部分 链接与backref的关系 .
多对一关系的反面总是 one to many . 可用的完整目录 relationship()
配置位于 基本关系模式 .
两种互补关系 Address.user
和 User.addresses
被称为 bidirectional relationship 是SQLAlchemy ORM的一个关键特性。断面 链接与backref的关系 详细讨论了“backref”特性。
论据 relationship()
它涉及到远程类,可以使用字符串指定,前提是声明性系统正在使用中。完成所有映射后,这些字符串将作为python表达式进行计算,以便生成实际参数,在上面的情况下, User
班级。在这个评估过程中允许的名称,除其他外,还包括根据声明的基创建的所有类的名称。
请参阅文档字符串 relationship()
有关参数样式的详细信息。
Did you know ?
大多数(但不是全部)关系数据库中的外键约束只能链接到主键列或具有唯一约束的列。
引用多列主键的外键约束本身具有多列,称为“复合外键”。它还可以引用这些列的子集。
外键列可以根据被引用列或行中的更改自动更新自己。这就是所谓的瀑布 引用操作 是关系数据库的内置函数。
外键可以引用自己的表。这被称为“自引用”外键。
有关外键的更多信息,请访问 Foreign Key - Wikipedia .
我们需要创建 addresses
数据库中的表,因此我们将从元数据中发出另一个create,它将跳过已创建的表:
sql>>> Base.metadata.create_all(engine)
PRAGMA...
CREATE TABLE addresses (
id INTEGER NOT NULL,
email_address VARCHAR NOT NULL,
user_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES users (id)
)
()
COMMIT
既然我们有两张表格,我们可以展示 Query
,特别是如何创建同时处理两个表的查询。这个 Wikipedia page on SQL JOIN 提供了一个关于连接技术的很好的介绍,其中一些我们将在这里进行说明。
在 User
和 Address
我们可以使用 Query.filter()
使它们的相关列相等。下面我们加载 User
和 Address
立即使用此方法的实体:
sql>>> for u, a in session.query(User, Address).\
... filter(User.id==Address.user_id).\
... filter(Address.email_address=='jack@google.com').\
... all():
... print(u)
... print(a)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname,
addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM users, addresses
WHERE users.id = addresses.user_id
AND addresses.email_address = ?
('jack@google.com',)
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
<Address(email_address='jack@google.com')>
另一方面,使用 Query.join()
方法:
sql>>> session.query(User).join(Address).\
... filter(Address.email_address=='jack@google.com').\
... all()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users JOIN addresses ON users.id = addresses.user_id
WHERE addresses.email_address = ?
('jack@google.com',)
[<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>]
Query.join()
知道如何加入 User
和 Address
因为它们之间只有一个外键。如果没有外带钥匙,或者有几把, Query.join()
当使用下列表单之一时效果更好:
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
正如您所期望的,同样的想法也用于“外部”连接,使用 outerjoin()
功能:
query.outerjoin(User.addresses) # LEFT OUTER JOIN
参考文件 join()
包含此方法接受的调用样式的详细信息和示例; join()
对于任何SQL Fluent应用程序来说,都是一种重要的使用中心方法。
什么? Query
如果存在多个实体,请从中选择?
这个 Query.join()
方法意志 通常从最左边的项联接 在实体列表中,如果省略了ON子句,或者ON子句是纯SQL表达式。要控制联接列表中的第一个实体,请使用 Query.select_from()
方法:
query = session.query(User, Address).select_from(Address).join(User)
当跨多个表进行查询时,如果同一个表需要被多次引用,则SQL通常要求该表 混叠的 使用另一个名称,以便将其与该表的其他出现项区分开来。这个 Query
使用 aliased
构建。下面我们加入 Address
实体两次,以查找同时具有两个不同电子邮件地址的用户:
>>> from sqlalchemy.orm import aliased
>>> adalias1 = aliased(Address)
>>> adalias2 = aliased(Address)
sql>>> for username, email1, email2 in \
... session.query(User.name, adalias1.email_address, adalias2.email_address).\
... join(adalias1, User.addresses).\
... join(adalias2, User.addresses).\
... filter(adalias1.email_address=='jack@google.com').\
... filter(adalias2.email_address=='j25@yahoo.com'):
... print(username, email1, email2)
SELECT users.name AS users_name,
addresses_1.email_address AS addresses_1_email_address,
addresses_2.email_address AS addresses_2_email_address
FROM users JOIN addresses AS addresses_1
ON users.id = addresses_1.user_id
JOIN addresses AS addresses_2
ON users.id = addresses_2.user_id
WHERE addresses_1.email_address = ?
AND addresses_2.email_address = ?
('jack@google.com', 'j25@yahoo.com')
jack jack@google.com j25@yahoo.com
这个 Query
适用于生成可以用作子查询的语句。假设我们想加载 User
对象以及数量 Address
每个用户拥有的记录。生成这样的SQL的最佳方法是获取按用户ID分组的地址计数,并加入到父级。在本例中,我们使用左外部联接,以便为没有地址的用户返回行,例如:
SELECT users.*, adr_count.address_count FROM users LEFT OUTER JOIN
(SELECT user_id, count(*) AS address_count
FROM addresses GROUP BY user_id) AS adr_count
ON users.id=adr_count.user_id
使用 Query
我们从内到外构建这样的语句。这个 statement
访问器返回表示特定语句生成的语句的SQL表达式 Query
-这是一个 select()
构造,如 SQL表达式语言教程 ::
>>> from sqlalchemy.sql import func
>>> stmt = session.query(Address.user_id, func.count('*').\
... label('address_count')).\
... group_by(Address.user_id).subquery()
这个 func
关键字生成SQL函数,并且 subquery()
方法对 Query
生成一个SQL表达式构造,表示嵌入在别名中的select语句(它实际上是 query.statement.alias()
)
一旦我们有了声明,它就表现得像 Table
构造,例如我们为其创建的 users
在本教程的开头。语句中的列可以通过名为 c
:
sql>>> for u, count in session.query(User, stmt.c.address_count).\
... outerjoin(stmt, User.id==stmt.c.user_id).order_by(User.id):
... print(u, count)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname,
anon_1.address_count AS anon_1_address_count
FROM users LEFT OUTER JOIN
(SELECT addresses.user_id AS user_id, count(?) AS address_count
FROM addresses GROUP BY addresses.user_id) AS anon_1
ON users.id = anon_1.user_id
ORDER BY users.id
('*',)
<User(name='ed', fullname='Ed Jones', nickname='eddie')> None
<User(name='wendy', fullname='Wendy Williams', nickname='windy')> None
<User(name='mary', fullname='Mary Contrary', nickname='mary')> None
<User(name='fred', fullname='Fred Flintstone', nickname='freddy')> None
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')> 2
上面,我们刚刚选择了一个包含子查询中的列的结果。如果我们希望子查询映射到一个实体呢?为此我们使用 aliased()
要将映射类的“别名”与子查询关联,请执行以下操作:
sql>>> stmt = session.query(Address).\
... filter(Address.email_address != 'j25@yahoo.com').\
... subquery()
>>> adalias = aliased(Address, stmt)
>>> for user, address in session.query(User, adalias).\
... join(adalias, User.addresses):
... print(user)
... print(address)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname,
anon_1.id AS anon_1_id,
anon_1.email_address AS anon_1_email_address,
anon_1.user_id AS anon_1_user_id
FROM users JOIN
(SELECT addresses.id AS id,
addresses.email_address AS email_address,
addresses.user_id AS user_id
FROM addresses
WHERE addresses.email_address != ?) AS anon_1
ON users.id = anon_1.user_id
('j25@yahoo.com',)
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
<Address(email_address='jack@google.com')>
SQL中的exists关键字是一个布尔运算符,如果给定表达式包含任何行,则返回true。它可以在许多场景中代替联接,也可以用于定位相关表中没有相应行的行。
有一个显式的exists构造,如下所示:
>>> from sqlalchemy.sql import exists
>>> stmt = exists().where(Address.user_id==User.id)
sql>>> for name, in session.query(User.name).filter(stmt):
... print(name)
SELECT users.name AS users_name
FROM users
WHERE EXISTS (SELECT *
FROM addresses
WHERE addresses.user_id = users.id)
()
jack
这个 Query
具有多个自动使用exists的运算符。上面的语句可以通过 User.addresses
关系使用 any()
:
sql>>> for name, in session.query(User.name).\
... filter(User.addresses.any()):
... print(name)
SELECT users.name AS users_name
FROM users
WHERE EXISTS (SELECT 1
FROM addresses
WHERE users.id = addresses.user_id)
()
jack
any()
也采用标准,以限制匹配的行:
sql>>> for name, in session.query(User.name).\
... filter(User.addresses.any(Address.email_address.like('%google%'))):
... print(name)
SELECT users.name AS users_name
FROM users
WHERE EXISTS (SELECT 1
FROM addresses
WHERE users.id = addresses.user_id AND addresses.email_address LIKE ?)
('%google%',)
jack
has()
是同一个运算符吗 any()
对于多对一关系(注意 ~
这里也有运算符,意思是“not”):
sql>>> session.query(Address).\
... filter(~Address.user.has(User.name=='jack')).all()
SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses
WHERE NOT (EXISTS (SELECT 1
FROM users
WHERE users.id = addresses.user_id AND users.name = ?))
('jack',)
[]
以下是所有建立在关系基础上的运算符-每个运算符都链接到其API文档,其中包含有关用法和行为的完整详细信息:
__eq__()
(多对一“等于”比较)::
query.filter(Address.user == someuser)
__ne__()
(多对一“不等于”比较)::
query.filter(Address.user != someuser)
为空(多对一比较,也使用 __eq__()
):
query.filter(Address.user == None)
contains()
(用于一对多集合)::
query.filter(User.addresses.contains(someaddress))
any()
(用于收藏)::
query.filter(User.addresses.any(Address.email_address == 'bar'))
# also takes keyword arguments:
query.filter(User.addresses.any(email_address='bar'))
has()
(用于标量引用)::
query.filter(Address.user.has(name='ed'))
Query.with_parent()
(用于任何关系)::
session.query(Address).with_parent(someuser, 'addresses')
回想一下之前我们画的 lazy loading 操作,当我们访问 User.addresses
A的集合 User
并发出SQL。如果您想减少查询的数量(在许多情况下,很大程度上),我们可以应用 eager load 到查询操作。SQLAlchemy提供三种类型的预加载,其中两种是自动加载,第三种是涉及自定义条件的加载。这三个函数通常都是通过以下函数调用的 query options 它给 Query
关于如何通过 Query.options()
方法。
在这种情况下,我们想指出 User.addresses
应该很快装载。加载一组对象及其相关集合的好选择是 orm.selectinload()
选项,它发出第二条select语句,完全加载与刚刚加载的结果关联的集合。“select in”的名称源自这样一个事实,即select语句使用in子句一次查找多个对象的相关行:
>>> from sqlalchemy.orm import selectinload
sql>>> jack = session.query(User).\
... options(selectinload(User.addresses)).\
... filter_by(name='jack').one()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ?
('jack',)
SELECT addresses.user_id AS addresses_user_id,
addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address
FROM addresses
WHERE addresses.user_id IN (?)
ORDER BY addresses.user_id, addresses.id
(5,)
>>> jack
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
另一个自动预加载函数更为著名,它被称为 orm.joinedload()
. 这种类型的加载会发出一个联接,默认情况下是一个左外部联接,以便在一个步骤中加载前导对象以及相关对象或集合。我们举例说明加载相同的 addresses
以这种方式收集-请注意,即使 User.addresses
收集关于 jack
现在已实际填充,查询将发出额外的联接,无论:
>>> from sqlalchemy.orm import joinedload
sql>>> jack = session.query(User).\
... options(joinedload(User.addresses)).\
... filter_by(name='jack').one()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname,
addresses_1.id AS addresses_1_id,
addresses_1.email_address AS addresses_1_email_address,
addresses_1.user_id AS addresses_1_user_id
FROM users
LEFT OUTER JOIN addresses AS addresses_1 ON users.id = addresses_1.user_id
WHERE users.name = ? ORDER BY addresses_1.id
('jack',)
>>> jack
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
注意,即使外部联接产生了两行,我们仍然只有一个 User
回来。这是因为 Query
对返回的实体应用基于对象标识的“统一”策略。这是为了在不影响查询结果的情况下应用连接的预先加载。
同时 joinedload()
已经很久了, selectinload()
是一种新的渴望装载形式。 selectinload()
倾向于更适合加载相关集合,而 joinedload()
倾向于更适合多对一的关系,因为只有一行同时为Lead和相关对象加载。另一种装载方式, subqueryload()
,也存在,可用于替代 selectinload()
在某些后端使用复合主键时。
joinedload()
is not a replacement for join()
由创建的联接 joinedload()
匿名化名,以便 不影响查询结果 . 安 Query.order_by()
或 Query.filter()
调用 不能 引用这些别名表-所谓的“用户空间”连接是使用 Query.join()
. 理由是 joinedload()
仅用于影响相关对象或集合作为优化详细信息的加载方式-它可以添加或删除,而不会影响实际结果。见剖面图 加入渴望装载的禅宗 有关如何使用它的详细说明。
第三种类型的抢先加载是当我们显式构造一个联接以定位主要行时,并且希望将额外的表应用于主要对象上的相关对象或集合。此功能通过 orm.contains_eager()
函数,对于在需要对同一对象进行筛选的查询上预加载多对一对象最为有用。下面我们演示加载 Address
行以及相关的 User
对象,筛选 User
命名为“杰克”,并使用 orm.contains_eager()
将“用户”列应用于 Address.user
属性:
>>> from sqlalchemy.orm import contains_eager
sql>>> jacks_addresses = session.query(Address).\
... join(Address.user).\
... filter(User.name=='jack').\
... options(contains_eager(Address.user)).\
... all()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname,
addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses JOIN users ON users.id = addresses.user_id
WHERE users.name = ?
('jack',)
>>> jacks_addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]
>>> jacks_addresses[0].user
<User(name='jack', fullname='Jack Bean', nickname='gjffdd')>
有关预加载的详细信息,包括如何在默认情况下配置各种加载形式,请参阅部分 关系加载技术 .
让我们尝试删除 jack
看看情况如何。我们将在会话中将对象标记为已删除,然后发出 count
查询以查看是否没有保留行:
>>> session.delete(jack)
sql>>> session.query(User).filter_by(name='jack').count()
UPDATE addresses SET user_id=? WHERE addresses.id = ?
((None, 1), (None, 2))
DELETE FROM users WHERE users.id = ?
(5,)
SELECT count(*) AS count_1
FROM (SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ?) AS anon_1
('jack',)
0
到目前为止,一切都很好。杰克的怎么样 Address
对象?
sql>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (?, ?)) AS anon_1
('jack@google.com', 'j25@yahoo.com')
2
噢,他们还在那儿!分析flush SQL,我们可以看到 user_id
每个地址的列设置为空,但未删除行。sqlAlchemy不假定删除层叠,您必须告诉它这样做。
我们将配置 叶栅 选项 User.addresses
改变行为的关系。虽然sqlAlchemy允许您在任何时间点向映射添加新的属性和关系,但在这种情况下,需要删除现有的关系,因此我们需要完全删除映射并重新开始-我们将关闭 Session
::
>>> session.close()
ROLLBACK
使用新的 declarative_base()
::
>>> Base = declarative_base()
下一步我们要申报 User
类,添加到 addresses
包括级联配置的关系(我们也将不考虑构造函数)::
>>> class User(Base):
... __tablename__ = 'users'
...
... id = Column(Integer, primary_key=True)
... name = Column(String)
... fullname = Column(String)
... nickname = Column(String)
...
... addresses = relationship("Address", back_populates='user',
... cascade="all, delete, delete-orphan")
...
... def __repr__(self):
... return "<User(name='%s', fullname='%s', nickname='%s')>" % (
... self.name, self.fullname, self.nickname)
然后我们重新创造 Address
注意到在这种情况下,我们已经创建了 Address.user
关系通过 User
类已经:
>>> class Address(Base):
... __tablename__ = 'addresses'
... id = Column(Integer, primary_key=True)
... email_address = Column(String, nullable=False)
... user_id = Column(Integer, ForeignKey('users.id'))
... user = relationship("User", back_populates="addresses")
...
... def __repr__(self):
... return "<Address(email_address='%s')>" % self.email_address
现在当我们加载用户时 jack
(以下使用 get()
,按主键加载),从相应的 addresses
收集将导致 Address
被删除:
# load Jack by primary key
sql>>> jack = session.query(User).get(5)
BEGIN (implicit)
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.id = ?
(5,)
# remove one Address (lazy load fires off)
sql>>> del jack.addresses[1]
SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses
WHERE ? = addresses.user_id
(5,)
# only one address remains
sql>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
DELETE FROM addresses WHERE addresses.id = ?
(2,)
SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (?, ?)) AS anon_1
('jack@google.com', 'j25@yahoo.com')
1
删除Jack将同时删除Jack和其余的 Address
与用户关联:
>>> session.delete(jack)
sql>>> session.query(User).filter_by(name='jack').count()
DELETE FROM addresses WHERE addresses.id = ?
(1,)
DELETE FROM users WHERE users.id = ?
(5,)
SELECT count(*) AS count_1
FROM (SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ?) AS anon_1
('jack',)
0
sql>>> session.query(Address).filter(
... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (?, ?)) AS anon_1
('jack@google.com', 'j25@yahoo.com')
0
我们将进入奖金回合,但让我们展示一种多对多的关系。我们还将潜入一些其他功能,只是为了参观一下。我们将使我们的应用程序成为一个博客应用程序,用户可以在其中编写 BlogPost
项目,其中 Keyword
与其关联的项。
对于普通的多对多,我们需要创建一个未映射的 Table
构造以用作关联表。如下所示:
>>> from sqlalchemy import Table, Text
>>> # association table
>>> post_keywords = Table('post_keywords', Base.metadata,
... Column('post_id', ForeignKey('posts.id'), primary_key=True),
... Column('keyword_id', ForeignKey('keywords.id'), primary_key=True)
... )
上面,我们可以看到声明 Table
直接与声明映射类稍有不同。 Table
是一个构造函数函数,因此每个 Column
参数用逗号分隔。这个 Column
对象也被显式地赋予其名称,而不是从分配的属性名称中获取。
下一步我们定义 BlogPost
和 Keyword
,使用互补 relationship()
构造,每个引用 post_keywords
表作为关联表::
>>> class BlogPost(Base):
... __tablename__ = 'posts'
...
... id = Column(Integer, primary_key=True)
... user_id = Column(Integer, ForeignKey('users.id'))
... headline = Column(String(255), nullable=False)
... body = Column(Text)
...
... # many to many BlogPost<->Keyword
... keywords = relationship('Keyword',
... secondary=post_keywords,
... back_populates='posts')
...
... def __init__(self, headline, body, author):
... self.author = author
... self.headline = headline
... self.body = body
...
... def __repr__(self):
... return "BlogPost(%r, %r, %r)" % (self.headline, self.body, self.author)
>>> class Keyword(Base):
... __tablename__ = 'keywords'
...
... id = Column(Integer, primary_key=True)
... keyword = Column(String(50), nullable=False, unique=True)
... posts = relationship('BlogPost',
... secondary=post_keywords,
... back_populates='keywords')
...
... def __init__(self, keyword):
... self.keyword = keyword
注解
上面的类声明说明了显式的 __init__()
方法。记住,当使用声明性时,它是可选的!
上面,多对多关系是 BlogPost.keywords
. 多对多关系的定义特征是 secondary
关键字参数引用 Table
表示关联表的对象。此表只包含引用关系两侧的列;如果它包含 any 其他列,如它自己的主键或其他表的外键,SQLAlchemy需要一个称为“关联对象”的不同使用模式,如 关联对象 .
我们也想要我们的 BlogPost
类有一个 author
字段。我们将把它添加为另一个双向关系,除了一个问题,我们将有一个单一的用户可能有很多博客文章。当我们进入 User.posts
,我们希望能够进一步筛选结果,以便不加载整个集合。为此,我们使用 relationship()
调用 lazy='dynamic'
,用于配置替代项 装载机策略 在属性上:
>>> BlogPost.author = relationship(User, back_populates="posts")
>>> User.posts = relationship(BlogPost, back_populates="author", lazy="dynamic")
创建新表:
sql>>> Base.metadata.create_all(engine)
PRAGMA...
CREATE TABLE keywords (
id INTEGER NOT NULL,
keyword VARCHAR(50) NOT NULL,
PRIMARY KEY (id),
UNIQUE (keyword)
)
()
COMMIT
CREATE TABLE posts (
id INTEGER NOT NULL,
user_id INTEGER,
headline VARCHAR(255) NOT NULL,
body TEXT,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES users (id)
)
()
COMMIT
CREATE TABLE post_keywords (
post_id INTEGER NOT NULL,
keyword_id INTEGER NOT NULL,
PRIMARY KEY (post_id, keyword_id),
FOREIGN KEY(post_id) REFERENCES posts (id),
FOREIGN KEY(keyword_id) REFERENCES keywords (id)
)
()
COMMIT
用法与我们所做的没有太大的不同。让我们给温迪一些博客文章:
sql>>> wendy = session.query(User).\
... filter_by(name='wendy').\
... one()
SELECT users.id AS users_id,
users.name AS users_name,
users.fullname AS users_fullname,
users.nickname AS users_nickname
FROM users
WHERE users.name = ?
('wendy',)
>>> post = BlogPost("Wendy's Blog Post", "This is a test", wendy)
>>> session.add(post)
我们在数据库中唯一地存储关键字,但是我们知道我们还没有关键字,所以我们可以创建它们:
>>> post.keywords.append(Keyword('wendy'))
>>> post.keywords.append(Keyword('firstpost'))
我们现在可以用关键字“firstpost”查找所有的博客文章。我们将使用 any
操作符查找“博客文章,其中任何关键字都包含关键字字符串‘firstpost’”:
sql>>> session.query(BlogPost).\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
INSERT INTO keywords (keyword) VALUES (?)
('wendy',)
INSERT INTO keywords (keyword) VALUES (?)
('firstpost',)
INSERT INTO posts (user_id, headline, body) VALUES (?, ?, ?)
(2, "Wendy's Blog Post", 'This is a test')
INSERT INTO post_keywords (post_id, keyword_id) VALUES (?, ?)
(...)
SELECT posts.id AS posts_id,
posts.user_id AS posts_user_id,
posts.headline AS posts_headline,
posts.body AS posts_body
FROM posts
WHERE EXISTS (SELECT 1
FROM post_keywords, keywords
WHERE posts.id = post_keywords.post_id
AND keywords.id = post_keywords.keyword_id
AND keywords.keyword = ?)
('firstpost',)
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]
如果我们想查找用户拥有的帖子 wendy
,我们可以告诉查询缩小到这个范围。 User
对象作为父级:
sql>>> session.query(BlogPost).\
... filter(BlogPost.author==wendy).\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
SELECT posts.id AS posts_id,
posts.user_id AS posts_user_id,
posts.headline AS posts_headline,
posts.body AS posts_body
FROM posts
WHERE ? = posts.user_id AND (EXISTS (SELECT 1
FROM post_keywords, keywords
WHERE posts.id = post_keywords.post_id
AND keywords.id = post_keywords.keyword_id
AND keywords.keyword = ?))
(2, 'firstpost')
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]
或者我们可以用温迪自己的 posts
关系,这是一种"动态"关系,直接从中查询:
sql>>> wendy.posts.\
... filter(BlogPost.keywords.any(keyword='firstpost')).\
... all()
SELECT posts.id AS posts_id,
posts.user_id AS posts_user_id,
posts.headline AS posts_headline,
posts.body AS posts_body
FROM posts
WHERE ? = posts.user_id AND (EXISTS (SELECT 1
FROM post_keywords, keywords
WHERE posts.id = post_keywords.post_id
AND keywords.id = post_keywords.keyword_id
AND keywords.keyword = ?))
(2, 'firstpost')
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', nickname='windy')>)]