版本:1.1.0b2 |发布日期:2016年7月1日

SQLAlchemy 1.1文档

对象关系教程

SQLAlchemy ORM提供了将用户定义的Python类和数据库表以及这些类(对象)的实例与其对应表中的行相关联的方法。它包括一个透明地同步对象及其相关行之间状态的所有更改的系统,称为工作单元,以及用于根据用户定义的类及其定义的表达式表达数据库查询的系统彼此之间的关系。

ORM与构建ORM的SQLAlchemy表达式语言形成鲜明对比。鉴于在SQL表达式语言教程中引入的SQL表达式语言提供了直接表示关系数据库的原始结构的系统,而没有意见,ORM呈现高级别和抽象的使用模式,是表达式语言的应用使用的示例。

虽然ORM和表达式语言的使用模式之间存在重叠,但是它们的相似之处却比最初出现时更为肤浅。从用户定义的域模型的角度来看,数据的结构和内容是透明持久化的,并从其底层存储模型刷新。另一种方法从文字模式和SQL表达式表达式的角度来看,它们被显式地组合成数据库单独消费的消息。

可以仅使用对象关系映射器来构建成功的应用。在高级情况下,使用ORM构建的应用程序可能会在需要特定数据库交互的某些区域中直接偶尔使用表达式语言。

The following tutorial is in doctest format, meaning each >>> line represents something you can type at a Python command prompt, and the following text represents the expected return value.

版本检查

快速检查以确认我们至少处于SQLAlchemy的版本1.1

>>> import sqlalchemy
>>> sqlalchemy.__version__ 
1.1.0

连接¶ T0>

对于本教程,我们将使用一个仅内存的SQLite数据库。要连接,请使用create_engine()

>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite:///:memory:', echo=True)

echo标志是设置SQLAlchemy日志记录的快捷方式,它是通过Python的标准日志模块完成的。启用它,我们将看到生成的所有SQL。如果您正在学习本教程并希望产生更少的输出,请将其设置为False本教程将把SQL格式化为一个弹出窗口,所以不会妨碍我们;只需点击“SQL”链接即可查看正在生成的内容。

create_engine()的返回值是Engine的一个实例,它表示数据库的核心接口,通过方言数据库和DBAPI的使用细节。在这种情况下,SQLite方言会将指令解释为Python内置的sqlite3模块。

第一次调用Engine.execute()Engine.connect()的方法时,Engine才会建立一个真实的DBAPI 连接到数据库,然后用于发出SQL。使用ORM时,我们通常不会直接使用Engine;相反,它在后台被ORM使用,我们将很快看到。

也可以看看

数据库网址 - 包括连接到多种数据库的create_engine()示例,其中包含指向更多信息的链接。

声明映射

当使用ORM时,配置过程首先描述我们将要处理的数据库表,然后定义我们用来映射到那些表的类。在现代SQLAlchemy中,这两个任务通常使用称为Declarative方法一起执行,这允许我们创建包含指令的类来描述它们将被映射到的实际数据库表。

使用Declarative方法定义的映射类依据一个基类,这个基类是维系类和数据表关系的目录 - 我们说说的Declarative base class在一个普通的模块入口中,应用通常只需要有一个base的实例。我们使用declarative_base()函数创建基类,如下所示:

>>> from sqlalchemy.ext.declarative import declarative_base

>>> Base = declarative_base()

现在我们有一个“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)
...     password = Column(String)
...
...     def __repr__(self):
...        return "<User(name='%s', fullname='%s', password='%s')>" % (
...                             self.name, self.fullname, self.password)

使用Declarative的类至少需要一个__ tablename __属性和至少一个Column,它是主键[1]的一部分。SQLAlchemy从不对类引用的表本身做任何假设,包括它没有内置的名称,数据类型或约束的约定。但这并不意味着需要样板;而是鼓励使用辅助函数和mixin类创建自己的自动约定,这在Mixin和Custom Base Classes中有详细描述。

构造类时,Declarative会使用称为描述符的特殊Python访问器替换所有Column对象;这是一个称为instrumentation的过程。“instrumented”映射类将为我们提供在SQL上下文中引用我们的表的方法,以及从数据库中持久化和加载列的值。

除了映射过程对我们的类做的外,类另外主要是一个普通的Python类,我们可以定义任何数量的普通属性和我们的应用程序所需的方法。

[1] T0>有关为什么需要主键的信息,请参阅如何映射没有主键的表?

构造模式

使用通过声明式系统构建的User类,我们定义了有关表的信息,称为表元数据SQLAlchemy用于表示特定表的此信息的对象称为对象,这里Declarative已经为我们做了一个。我们可以通过检查__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('password', String(), table=<users>), schema=None)

When we declared our class, Declarative used a Python metaclass in order to perform additional activities once the class declaration was complete; within this phase, it then created a Table object according to our specifications, and associated it with the class by constructing a Mapper object. 这个对象是我们通常不需要直接处理的幕后对象(虽然在我们需要的时候它可以提供关于我们映射的大量信息)。

Table对象是一个更大集合的成员,被称为MetaData当使用Declarative时,这个对象可以使用我们的声明基类的.metadata属性。

The MetaData is a registry which includes the ability to emit a limited set of schema generation commands to the database. 由于我们的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,
    password VARCHAR,
    PRIMARY KEY (id)
)
()
COMMIT

最小表格描述与完整描述

熟悉CREATE TABLE语法的用户可能会注意到VARCHAR列的生成没有长度;在SQLite和Postgresql上,这是一个有效的数据类型,但在其他情况下,这是不允许的。因此,如果在其中一个数据库上运行本教程,并且希望使用SQLAlchemy发出CREATE TABLE,则可以为String类型提供“length”,如下所示:

Column(String(50))

String上的长度字段以及IntegerNumeric等可用的类似精度/缩放字段。除了创建表时,不会被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))
    password = Column(String(12))

    def __repr__(self):
        return "<User(name='%s', fullname='%s', password='%s')>" % (
                                self.name, self.fullname, self.password)

我们单独包含这个更详细的表定义,以突出显示主要针对Python内使用的最小构造与将用于在具有更严格要求的特定后端组上发出CREATE TABLE语句的构造之间的差异。

创建映射类的实例

完成映射后,我们现在创建并检查User对象:

>>> ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')
>>> ed_user.name
'ed'
>>> ed_user.password
'edspassword'
>>> str(ed_user.id)
'None'

Even though we didn’t specify it in the constructor, the id attribute still produces a value of None when we access it (as opposed to Python’s usual behavior of raising AttributeError for an undefined attribute). SQLAlchemy的instrumentation通常在首次访问时为列映射属性生成此默认值。对于那些我们实际赋值的属性,工具系统正在跟踪这些赋值,以便在最终的INSERT语句中使用,以便发送到数据库。

创建一个会话

我们现在准备开始与数据库会话了。ORM通过Session与数据库建立连接的。当我们首次启动应用程序时,与我们的create_engine()语句相同,我们定义一个Session类,作为新的Session 对象:

>>> from sqlalchemy.orm import sessionmaker
>>> Session = sessionmaker(bind=engine)

在您的应用程序在定义模块级对象时尚未具有Engine的情况下,只需按如下设置:

>>> Session = sessionmaker()

稍后,当您使用create_engine()创建引擎时,请使用configure()将其连接到Session

>>> 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', password='edspassword')
>>> session.add(ed_user)

此时,我们说这个实例正在等待;尚未发出SQL,并且对象尚未由数据库中的行表示。根据需要,Session将会发出SQL来坚持Ed tt4> Jones 冲洗 T6>。如果我们在数据库中查询Ed Jones,所有待处理的信息将首先被刷新,查询立即发出。

例如,下面我们创建一个新的Query对象,它加载User的实例。我们“过滤”edname属性,并指示我们只想获取完整列表中的第一个结果。返回一个与我们添加的相同的User实例:

sql>>> our_user = session.query(User).filter_by(name='ed').first() # doctest:+NORMALIZE_WHITESPACE
>>> our_user
<User(name='ed', fullname='Ed Jones', password='edspassword')>

事实上,Session已经识别出返回的行是与其内部对象映射中已经表示的相同的行,因此我们实际上得到了与我们刚才添加的相同的实例:

>>> ed_user is our_user
True

这里工作的ORM概念被称为身份映射并且确保在Session内的特定行上的所有操作对相同的数据集进行操作。一旦具有特定主键的对象出现在Session中,那么会话上的所有SQL查询将始终为该特定主键返回相同的Python对象;如果试图在会话中放置具有相同主键的第二个已经持久化的对象,它也将引发错误。

我们可以使用add_all()一次添加更多User对象:

>>> session.add_all([
...     User(name='wendy', fullname='Wendy Williams', password='foobar'),
...     User(name='mary', fullname='Mary Contrary', password='xxg527'),
...     User(name='fred', fullname='Fred Flinstone', password='blah')])

此外,我们认为Ed的密码不是太安全,所以让我们改变它:

>>> ed_user.password = 'f8s7ccs'

Session正在关注。例如,它知道Ed Jones已被修改:

>>> session.dirty
IdentitySet([<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>])

并且三个新的User对象正在等待:

>>> session.new  # doctest: +SKIP
IdentitySet([<User(name='wendy', fullname='Wendy Williams', password='foobar')>,
<User(name='mary', fullname='Mary Contrary', password='xxg527')>,
<User(name='fred', fullname='Fred Flinstone', password='blah')>])

我们告诉session,我们要发布对数据库的所有剩余更改,并提交已经在进行中的事务。我们通过commit()来实现。session发出用于“ed”上的密码更改的UPDATE语句,以及三个新的User语句INSERT我们添加的对象:

sql>>> session.commit()

commit()刷新数据库中剩余的任何更改,并提交事务。session引用的连接资源现在返回到连接池。此会话的后续操作将发生在事务中,这将在第一次需要时再次重新获取连接资源。

如果我们看一下Ed的id属性,它早先是None,它现在有一个值:

sql>>> ed_user.id # doctest: +NORMALIZE_WHITESPACE
1

会话在数据库中插入新行后,所有新生成的标识符和数据库生成的默认值立即或通过首次加载访问在实例上可用。在这种情况下,整个行在访问时重新加载,因为在我们发出commit()之后开始一个新事务。SQLAlchemy默认情况下刷新第一次在新事务中访问之前的事务的数据,以便最近的状态可用。重新加载的级别是可以配置的,如使用session中所述。

会话对象状态

由于我们的用户对象从会话之外移动到没有主键的会话内部,为了实际插入,它在三暂停等待持久性中的四个可用的“对象状态”。意识到这些状态,它们的含义总是一个好主意 - 请务必阅读Quickie简介到对象状态以便快速了解。

回滚

由于Session在一个事务中工作,因此我们可以回滚更改。让我们做将被还原的两个更改; ed_user的用户名设置为Edwardo

>>> ed_user.name = 'Edwardo'

我们添加另一个错误的用户fake_user

>>> fake_user = User(name='fakeuser', fullname='Invalid', password='12345')
>>> session.add(fake_user)

查询会话,我们可以看到它们被刷入当前事务:

sql>>> session.query(User).filter(User.name.in_(['Edwardo', 'fakeuser'])).all()
[<User(name='Edwardo', fullname='Ed Jones', password='f8s7ccs')>, <User(name='fakeuser', fullname='Invalid', password='12345')>]

回滚,我们可以看到ed_user的名称已回到ed,而且fake_user已被踢出会话:

sql>>> session.rollback()

sql>>> ed_user.name
u'ed'
>>> fake_user in session
False

发出SELECT说明对数据库所做的更改:

sql>>> session.query(User).filter(User.name.in_(['ed', 'fakeuser'])).all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>]

查询

使用Session上的query()方法创建Query对象。此函数采用可变数量的参数,参数可以是类或者类的描述的集合。下面,我们指示加载User实例的查询下面是一个迭代输出User类的例子:

sql>>> for instance in session.query(User).order_by(User.id):
...     print(instance.name, instance.fullname)
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flinstone

Query也接受ORM描述作为参数。任何时候,多个类实体或基于列的实体表达都可以作为query()函数的参数,返回类型为元组:

sql>>> for name, fullname in session.query(User.name, User.fullname):
...     print(name, fullname)
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flinstone

Query返回的元组是命名为tuples,由KeyedTuple类提供,可以像普通Python对象一样对待。名称与属性的属性名称以及类的类名相同:

sql>>> for row in session.query(User, User.name).all():
...    print(row.User, row.name)
<User(name='ed', fullname='Ed Jones', password='f8s7ccs')> ed
<User(name='wendy', fullname='Wendy Williams', password='foobar')> wendy
<User(name='mary', fullname='Mary Contrary', password='xxg527')> mary
<User(name='fred', fullname='Fred Flinstone', password='blah')> fred

您可以使用label()结构控制单个列表达式的名称,该结构可从任何ColumnElement - 来源对象,以及任何映射到实体表的列元素(例如User.name):

sql>>> for row in session.query(User.name.label('name_label')).all():
...    print(row.name_label)
ed
wendy
mary
fred

假设在对query()的调用中存在多个实体,可以使用aliased()来控制完整实体例如User

>>> 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)
<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>
<User(name='wendy', fullname='Wendy Williams', password='foobar')>
<User(name='mary', fullname='Mary Contrary', password='xxg527')>
<User(name='fred', fullname='Fred Flinstone', password='blah')>

使用Query的基本操作包括发出LIMIT和OFFSET,最方便地使用Python数组分片,通常与ORDER BY结合使用:

sql>>> for u in session.query(User).order_by(User.id)[1:3]:
...    print(u)
<User(name='wendy', fullname='Wendy Williams', password='foobar')>
<User(name='mary', fullname='Mary Contrary', password='xxg527')>

过滤结果,可通过使用filter_by()(使用关键字参数)完成:

sql>>> for name, in session.query(User.name).\
...             filter_by(fullname='Ed Jones'):
...    print(name)
ed

...或filter(),它使用更灵活的SQL表达式语言结构。这些允许您在映射类上使用带有类级属性的常规Python运算符:

sql>>> for name, in session.query(User.name).\
...             filter(User.fullname=='Ed Jones'):
...    print(name)
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)
<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>

常用过滤器操作符

下面是filter()中使用的一些最常见的运算符:

  • equals

    query.filter(User.name == 'ed')
  • not equals

    query.filter(User.name != 'ed')
  • LIKE

    query.filter(User.name.like('%ed%'))
  • 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_()不是 Python and运算符!

  • OR

    from sqlalchemy import or_
    query.filter(or_(User.name == 'ed', User.name == 'wendy'))

注意

请确保使用or_()不是 Python or运算符!

  • MATCH

    query.filter(User.name.match('wendy'))

注意

match()使用数据库特定的MATCHCONTAINS函数;其行为将随后端而变化,并且在某些后端(如SQLite)上不可用。

返回列表和标量

Query上的多个方法会立即发出SQL,并返回包含已加载数据库结果的值。这里是一个简短的介绍:

  • all()返回一个列表:

    >>> query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)
    sql>>> query.all()
    [<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>,
          <User(name='fred', fullname='Fred Flinstone', password='blah')>]
  • first()应用一个限制,并将第一个结果作为标量返回:

    sql>>> query.first()
    <User(name='ed', fullname='Ed Jones', password='f8s7ccs')>
  • 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()方法非常适用于希望处理“没有找到项目”和“找到多个项目”的系统;例如Web服务返回,当没有找到结果时,它可能想要引发“404未找到”,但是在找到多个结果时引发应用程序错误。

  • one_or_none()就像one(),除了如果没有找到结果,它不会引发错误;它只返回Noneone(),如果找到多个结果,它会引发错误。

  • scalar()调用one()方法,在one()成功的基础上返回该行的第一列:

    >>> query = session.query(User.id).filter(User.name == 'ed').\
    ...    order_by(User.id)
    sql>>> query.scalar()
    1

使用文本SQL

文本字符串可以通过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)
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()
<User(name='fred', fullname='Fred Flinstone', password='blah')>

要使用一个完整的SQL语句,可以将表示完整语句的text()结构传递给from_statement()如果没有其他说明符,字符串SQL中的列将根据名称与模型列匹配,如下所示,我们只使用星号来表示加载所有列:

sql>>> session.query(User).from_statement(
...                     text("SELECT * FROM users where name=:name")).\
...                     params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>]

名称匹配列适用于简单情况,但在处理包含重复列名的复杂语句时或使用不易与特定名称匹配的匿名ORM构造时,可能会变得不方便。另外,在我们映射的列中存在打字行为,在处理结果行时我们可能会发现这一点。对于这些情况,text()结构允许我们将其文本SQL链接到Core或ORM映射的列表达式;我们可以通过将列表达式作为位置参数传递到TextClause.columns()方法来实现:

>>> stmt = text("SELECT name, id, fullname, password "
...             "FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id, User.fullname, User.password)
sql>>> session.query(User).from_statement(stmt).params(name='ed').all()
[<User(name='ed', fullname='Ed Jones', password='f8s7ccs')>]

版本1.1中的新功能: TextClause.columns()方法现在接受与明文SQL结果集位置匹配的列表达式,匹配或甚至在SQL语句中是唯一的。

When selecting from a text() construct, the Query may still specify what columns and entities are to be returned; instead of query(User) we can also ask for the columns individually, as in any other case:

>>> 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()
[(1, u'ed')]

也可以看看

Using Textual SQL - 从纯核查询的角度解释text()结构。

计数¶ T0>

Query includes a convenience method for counting called count():

sql>>> session.query(User).filter(User.name.like('%ed')).count()
2

The count() method is used to determine how many rows the SQL statement would return. 查看上面生成的SQL,SQLAlchemy总是将我们正在查询的任何东西放置到子查询中,然后对其进行计数。In some cases this can be reduced to a simpler SELECT count(*) FROM table, however modern versions of SQLAlchemy don’t try to guess when this is appropriate, as the exact SQL can be emitted using more explicit means.

For situations where the “thing to be counted” needs to be indicated specifically, we can specify the “count” function directly using the expression func.count(), available from the func construct. 下面我们用它来返回每个不同的用户名的计数:

>>> from sqlalchemy import func
sql>>> session.query(func.count(User.name), User.name).group_by(User.name).all()
[(1, u'ed'), (1, u'fred'), (1, u'mary'), (1, u'wendy')]

为了实现我们简单的SELECT count(*) FROM 它作为:

sql>>> session.query(func.count('*')).select_from(User).scalar()
4

如果直接使用User主键来表示计数,则可以删除select_from()

sql>>> session.query(func.count(User.id)).scalar()
4

建立关系

我们考虑如何映射和查询与User相关的第二张表。系统中的用户可以存储与其用户名相关联的任意数量的电子邮件地址。这意味着用户到存储电子邮件地址的新表(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的指令,表示该列中的值应该是受制于在User表中的值。这是关系数据库的一个核心功能,它是一个“粘合剂”,它可以转换不相连的表的集合,使其具有丰富的重叠关系。上面的ForeignKey表示addresses.user_id列中的值应该被限制在users.id列中的那些值上,也就是它的主键。

第二个指令称为relationship(),它告诉ORM:Address类本身应该链接到User类, Address.userrelationship() uses the foreign key relationships between the two tables to determine the nature of this linkage, determining that Address.user will be many to one. An additional relationship() directive is placed on the User mapped class under the attribute User.addresses. In both relationship() directives, the parameter relationship.back_populates is assigned to refer to the complementary attribute names; by doing so, each relationship() can make intelligent decision about the same relationship as expressed in reverse; on one side, Address.user refers to a User instance, and on the other side, User.addresses refers to a list of Address instances.

注意

relationship.back_populates参数是称为relationship.backref的非常常见的SQLAlchemy特性的更新版本。relationship.backref参数没有任何地方,并将始终保持可用!relationship.back_populates是相同的东西,除了稍微冗长一点,更容易操作。有关整个主题的概述,请参阅Linking Relationships with Backref部分。

多对一关系的反面总是one to manyrelationship()配置的完整目录位于Basic Relationship Patterns

两个互补关系Address.userUser.addresses被称为bidirectional relationship,并且是SQLAlchemy ORM的一个关键特性。The section Linking Relationships with Backref discusses the “backref” feature in detail.

关于远程类的relationship()的参数可以使用字符串指定,假定Declarative系统正在使用。一旦所有的映射完成,这些字符串被评估为Python表达式,以便产生实际的参数,在上面的例子中是User类。在评估过程中允许使用的名称除其他外,包括根据申报的基数创建的所有类别的名称。

有关参数样式的更多详细信息,请参阅relationship()的文档字符串。

你知道吗 ?

  • 大多数(尽管不是全部)关系数据库中的FOREIGN KEY约束只能链接到主键列或具有UNIQUE约束的列。
  • 引用多列主键的FOREIGN KEY约束,本身有多个列,被称为“组合外键”。它也可以引用这些列的一个子集。
  • FOREIGN KEY列可以自动更新自己,以响应引用的列或行的更改。这被称为CASCADE 引用操作,并且是关系数据库的内置函数。
  • FOREIGN KEY可以参考自己的表格。这被称为“自我参照”外键。
  • 外键 - 维基百科上阅读有关外键的更多信息。

我们需要在数据库中创建addresses表,所以我们将从元数据中发出另一个CREATE,这将跳过已经创建的表:

sql>>> Base.metadata.create_all(engine)

查询连接

Now that we have two tables, we can show some more features of Query, specifically how to create queries that deal with both tables at the same time. SQL JOIN上的Wikipedia页面提供了一个很好的介绍连接技术的方法,我们将在这里介绍其中的几个。

To construct a simple implicit join between User and Address, we can use Query.filter() to equate their related columns together. 下面我们使用这个方法一次加载UserAddress实体:

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)
<User(name='jack', fullname='Jack Bean', password='gjffdd')>
<Address(email_address='jack@google.com')>

另一方面,使用Query.join()方法最容易实现实际的SQL JOIN语法:

sql>>> session.query(User).join(Address).\
...         filter(Address.email_address=='jack@google.com').\
...         all()
[<User(name='jack', fullname='Jack Bean', password='gjffdd')>]

Query.join() knows how to join between User and Address because there’s only one foreign key between them. If there were no foreign keys, or several, Query.join() works better when one of the following forms are used:

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

The reference documentation for join() contains detailed information and examples of the calling styles accepted by this method; join() is an important method at the center of usage for any SQL-fluent application.

如果有多个实体,Query选择什么?

The Query.join() method will typically join from the leftmost item in the list of entities, when the ON clause is omitted, or if the ON clause is a plain SQL expression. 要控制JOIN列表中的第一个实体,请使用Query.select_from()方法:

query = Session.query(User, Address).select_from(Address).join(User)

使用别名

When querying across multiple tables, if the same table needs to be referenced more than once, SQL typically requires that the table be aliased with another name, so that it can be distinguished against other occurrences of that table. 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)
jack jack@google.com j25@yahoo.com

使用子查询

Query适用于生成可用作子查询的语句。假设我们想要载入User对象以及每个用户有多少个Address记录的计数。以这种方式生成SQL的最佳方法是获取按用户标识分组的地址数,并将其添加到父级。在这种情况下,我们使用LEFT OUTER JOIN,这样我们就得到了那些没有任何地址的用户的行,例如:

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访问器返回一个表达由特定Query生成的语句的SQL表达式 - 这是一个select()结构的实例,描述在SQL Expression Language Tutorial中:

>>> from sqlalchemy.sql import func
>>> stmt = session.query(Address.user_id, func.count('*').\
...         label('address_count')).\
...         group_by(Address.user_id).subquery()

The func keyword generates SQL functions, and the subquery() method on Query produces a SQL expression construct representing a SELECT statement embedded within an alias (it’s actually shorthand for 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)
<User(name='ed', fullname='Ed Jones', password='f8s7ccs')> None
<User(name='wendy', fullname='Wendy Williams', password='foobar')> None
<User(name='mary', fullname='Mary Contrary', password='xxg527')> None
<User(name='fred', fullname='Fred Flinstone', password='blah')> None
<User(name='jack', fullname='Jack Bean', password='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)
<User(name='jack', fullname='Jack Bean', password='gjffdd')>
<Address(email_address='jack@google.com')>

使用EXISTS

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)
jack

Query具有几个自动使用EXISTS的操作符。以上,语句可以使用any()沿User.addresses关系表示:

sql>>> for name, in session.query(User.name).\
...         filter(User.addresses.any()):
...     print(name)
jack

any() takes criterion as well, to limit the rows matched:

sql>>> for name, in session.query(User.name).\
...     filter(User.addresses.any(Address.email_address.like('%google%'))):
...     print(name)
jack

has() is the same operator as any() for many-to-one relationships (note the ~ operator here too, which means “NOT”):

sql>>> session.query(Address).\
...         filter(~Address.user.has(User.name=='jack')).all()
[]

普通关系运算符

以下是所有基于关系的运营商 - 每个运营商都链接到其API文档,其中包括使用情况和行为的详细信息:

  • __eq__()(多对一“等于”比较):

    query.filter(Address.user == someuser)
  • __ne__() (many-to-one “not equals” comparison):

    query.filter(Address.user != someuser)
  • IS NULL(多对一比较,也使用__eq__()):

    query.filter(Address.user == None)
  • contains()(用于一对多集合):

    query.filter(User.addresses.contains(someaddress))
  • any() (used for collections):

    query.filter(User.addresses.any(Address.email_address == 'bar'))
    
    # also takes keyword arguments:
    query.filter(User.addresses.any(email_address='bar'))
  • has() (used for scalar references):

    query.filter(Address.user.has(name='ed'))
  • Query.with_parent() (used for any relationship):

    session.query(Address).with_parent(someuser, 'addresses')

急于加载

Recall earlier that we illustrated a lazy loading operation, when we accessed the User.addresses collection of a User and SQL was emitted. 如果你想减少查询次数(在很多情况下),我们可以在查询操作中应用一个负载SQLAlchemy提供了三种急切加载类型,其中两种是自动加载的,另一种是涉及自定义标准的。All three are usually invoked via functions known as query options which give additional instructions to the Query on how we would like various attributes to be loaded, via the Query.options() method.

子查询Load

在这种情况下,我们希望指出User.addresses应该加载。加载一组对象及其相关集合的好的选择是orm.subqueryload()选项,该选项会发出第二个SELECT语句,以完全加载与刚加载的结果相关的集合。名称“subquery”源于直接通过Query构造的SELECT语句被重用,作为子查询嵌入到相关表的SELECT中。这是一个有点精心,但很容易使用:

>>> from sqlalchemy.orm import subqueryload
sql>>> jack = session.query(User).\
...                 options(subqueryload(User.addresses)).\
...                 filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', password='gjffdd')>

>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]

注意

subqueryload() when used in conjunction with limiting such as Query.first(), Query.limit() or Query.offset() should also include Query.order_by() on a unique column in order to ensure correct results. 请参阅The Importance of Ordering

加入Load

另一个自动加载的加载函数是比较有名的,叫做orm.joinedload()这种加载方式会发出一个JOIN,默认情况下是一个LEFT OUTER JOIN,这样一个步骤就可以加载前导对象以及相关的对象或集合。我们以这种方式说明加载相同的addresses集合 - 请注意,即使jack上的User.addresses集合现在实际上已被填充,查询将发出额外的连接,不管:

>>> from sqlalchemy.orm import joinedload

sql>>> jack = session.query(User).\
...                        options(joinedload(User.addresses)).\
...                        filter_by(name='jack').one()
>>> jack
<User(name='jack', fullname='Jack Bean', password='gjffdd')>

>>> jack.addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]

请注意,即使OUTER JOIN导致了两行,我们仍然只有一个User的实例。这是因为Query根据对象标识将“uniquing”策略应用于返回的实体。这是特别的,使得可以应用加入的加载加载而不影响查询结果。

While joinedload() has been around for a long time, subqueryload() is a newer form of eager loading. subqueryload() tends to be more appropriate for loading related collections while joinedload() tends to be better suited for many-to-one relationships, due to the fact that only one row is loaded for both the lead and the related object.

joinedload() is not a replacement for join()

joinedload()创建的连接是匿名化的,因此不影响查询结果Query.order_by()Query.filter()调用不能引用这些别名表 - 构建所谓的“用户空间”联接使用Query.join()这样做的基本原理是joinedload()仅用于影响如何将相关对象或集合作为优化细节加载 - 可以添加或删除它,而不影响实际结果。关于如何使用它的详细描述,请参见The Zen of Eager Loading一节。

显式加入+ Eagerload

第三种急切的加载方式是当我们显式地构造一个JOIN来定位主要的行,并且想额外地将额外的表应用到主要对象上的一个相关的对象或集合。此功能是通过orm.contains_eager()函数提供的,通常用于在需要过滤同一对象的查询上预加载多对一对象。下面我们举例加载一个Address行以及相关的User对象,过滤名为“jack”的User并使用orm.contains_eager()将“user”列应用于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()
>>> jacks_addresses
[<Address(email_address='jack@google.com')>, <Address(email_address='j25@yahoo.com')>]

>>> jacks_addresses[0].user
<User(name='jack', fullname='Jack Bean', password='gjffdd')>

有关加载的更多信息,包括如何配置各种形式的加载默认情况下,请参阅Relationship Loading Techniques部分。

删除¶ T0>

我们试着删除jack,看看结果如何。我们将在会话中标记为已删除,然后我们将发出一个count查询来查看没有行保留:

>>> session.delete(jack)
sql>>> session.query(User).filter_by(name='jack').count()
0

到现在为止还挺好。杰克的Address对象怎么样?

sql>>> session.query(Address).filter(
...     Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
...  ).count()
2

呃哦,他们还在那儿!通过分析flush SQL,我们可以看到每个地址的user_id列被设置为NULL,但是这些行没有被删除。SQLAlchemy不假定删除级联,你必须告诉它这样做。

配置delete / delete-orphan级联

我们将在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)
...     password = Column(String)
...
...     addresses = relationship("Address", back_populates='user',
...                     cascade="all, delete, delete-orphan")
...
...     def __repr__(self):
...        return "<User(name='%s', fullname='%s', password='%s')>" % (
...                                self.name, self.fullname, self.password)

然后重新创建Address,注意在这种情况下,我们已经通过User类创建了Address.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)

# remove one Address (lazy load fires off)
sql>>> del jack.addresses[1]

# only one address remains
sql>>> session.query(Address).filter(
...     Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
1

删除Jack将删除Jack和与用户关联的其余Address

>>> session.delete(jack)

sql>>> session.query(User).filter_by(name='jack').count()
0

sql>>> session.query(Address).filter(
...    Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
0

更多关于小瀑布

级联配置的进一步细节在CascadesThe cascade functionality can also integrate smoothly with the ON DELETE CASCADE functionality of the relational database. 有关详细信息,请参见Using Passive Deletes

建立一个多对多的关系

我们正在进入这个奖金轮,但让我们展示一个多对多的关系。我们也会潜入一些其他的功能,只是为了参观。我们将使我们的应用程序成为一个博客应用程序,用户可以在其中编写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 is a constructor function, so each individual Column argument is separated by a comma. Column对象也明确给出了它的名字,而不是从分配的属性名称中取出。

接下来,我们使用互补的relationship()结构定义BlogPostKeyword,每个结构都引用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__()方法。请记住,使用声明时,它是可选的!

Above, the many-to-many relationship is BlogPost.keywords. 多对多关系的定义特征是引用代表关联表的Table对象的secondary关键字参数。This table only contains columns which reference the two sides of the relationship; if it has any other columns, such as its own primary key, or foreign keys to other tables, SQLAlchemy requires a different usage pattern called the “association object”, described at Association Object.

我们还希望我们的BlogPost类有一个author字段。我们会将此添加为另一种双向关系,除了一个问题,我们将会有一个用户可能有大量的博客文章。当我们访问User.posts时,我们希望能够进一步过滤结果,以免加载整个集合。为此,我们使用一个名为lazy='dynamic'relationship()接受的设置,该设置在属性上配置一个替代的加载器策略

>>> BlogPost.author = relationship(User, back_populates="posts")
>>> User.posts = relationship(BlogPost, back_populates="author", lazy="dynamic")

创建新表格:

sql>>> Base.metadata.create_all(engine)

用法与我们所做的没有太大的不同。让我们给温蒂一些博客文章:

sql>>> wendy = session.query(User).\
...                 filter_by(name='wendy').\
...                 one()
>>> 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()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', password='foobar')>)]

如果我们想查找由用户wendy拥有的帖子,我们可以告诉查询缩小到User对象作为父级:

sql>>> session.query(BlogPost).\
...             filter(BlogPost.author==wendy).\
...             filter(BlogPost.keywords.any(keyword='firstpost')).\
...             all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', password='foobar')>)]

或者我们可以使用温迪自己的posts关系,这是一个“动态”关系,直接从那里查询:

sql>>> wendy.posts.\
...         filter(BlogPost.keywords.any(keyword='firstpost')).\
...         all()
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name='wendy', fullname='Wendy Williams', password='foobar')>)]

进一步参考

查询参考:query_api_toplevel

映射器参考:Mapper Configuration

关系参考:Relationship Configuration

会话参考:Using the Session