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

SQLAlchemy 1.1文档

ORM示例

SQLAlchemy发行版包含各种代码示例，说明一组选择的模式，一些是典型的，一些不太典型。所有都可以运行，可以在发行版的/examples目录中找到。所有的描述和源代码可以在这里找到。

额外的SQLAlchemy示例（某些用户贡献）可在维基上的http://www.sqlalchemy.org/trac/wiki/UsageRecipes上获得。

映射食谱

邻接列表

使用邻接列表模型映射的字典词典结构示例。

例如。：

node = TreeNode('rootnode')
node.append('node1')
node.append('node3')
session.add(node)
session.commit()

dump_tree(node)

文件清单：

协会¶ T0>

说明“关联对象”模式的用法的示例，其中中间类调解以多对多模式关联的两个类之间的关系。

文件清单：

  • proxied_association.py - 与basic_association相同的示例，在使用sqlalchemy.ext.associationproxy时添加以明确引用OrderItem可选。
  • basic_association.py - 说明“订单”和“商品”对象集合之间的多对多关系，通过名为“OrderItem”的关联对象将购买价格与每个对象关联起来
  • dict_of_sets_with_default.py - 一个高级关联代理示例，它演示了关联代理的嵌套以生成多级Python集合，在这种情况下，这是一个包含字符串键和整数集作为值的字典，它隐藏了映射的底层类。

有向图

有向图结构的持久性示例。该图存储为一组边，每个节都引用一个“下”节点和一个“上”节点。对低层和高层邻居的基本持久性和查询进行了说明：

n2 = Node(2)
n5 = Node(5)
n2.add_neighbor(n5)
print n2.higher_neighbors()

文件清单：

作为词典的动态关系

说明如何在“动态”关系之上放置类似字典的外观，以便字典操作（假设简单的字符串键）可以对大集合进行操作，而无需一次加载完整集合。

文件清单：

通用关联

举例说明将多种父母与特定子对象相关联的各种方法。

这些示例都使用声明性扩展和声明性混合。Each one presents the identical use case at the end - two classes, Customer and Supplier, both subclassing the HasAddresses mixin, which ensures that the parent class is provided with an addresses collection which contains Address objects.

该 discriminator_on_association.py T0>和 generic_fk.py T1>脚本现代化在2007年的博客文章多态关联与SQLAlchemy的 T2>提交食谱版本。

文件清单：

  • generic_fk.py - 以类似于Django，ROR等流行框架的方式说明所谓的“通用外键”。这种方法绕过了标准的参照完整性实践，因为“外键”列实际上并没有限制引用任何特定的表；相反，应用程序内部逻辑用于确定引用哪个表。
  • table_per_association.py - 说明一个混合，它通过为每个父类分别生成的关联表提供一个通用关联。关联的对象本身被保存在所有父母共享的单个表中。
  • discriminator_on_association.py - Illustrates a mixin which provides a generic association using a single target table and a single association table, referred to by all parent tables. 关联表包含一个“鉴别器”列，用于确定与关联表中每个特定行关联的父对象的类型。
  • table_per_related.py - 说明了一个通用关联，它在每个表内保存关联对象，每个关联对象都会生成以代表特定父类持久化这些对象。

大集合

大集合的例子。

说明当相关对象列表非常大时，与relationship()一起使用的选项，其中包括：

  • “动态”关系查询访问数据切片
  • 如何将ON DELETE CASCADE与passive_deletes=True结合使用，以大大提高相关集合删除的性能。

文件清单：

物化路径

使用SQLAlchemy ORM演示分层数据的“物化路径”模式。

文件清单：

嵌套集

说明使用SQLAlchemy ORM实现分层数据的“嵌套集”模式的基本方法。

文件清单：

性能¶ T0>

性能分析套件，适用于各种SQLAlchemy用例。

每个套件都专注于具有特定性能配置文件和相关含义的特定用例：

  • 批量插入
  • 个人插入，有或没有交易
  • 获取大量的行
  • 运行很多简短的查询

所有套件都包含了说明Core和ORM使用的各种使用模式，并且通常按照性能从最差到最大的顺序进行排序，与SQLAlchemy提供的功能数量相反，最大最小（这两种情况通常完全一致）。

在包级别提供了一个命令行工具，可以运行各个套件：

$ python -m examples.performance --help
usage: python -m examples.performance [-h] [--test TEST] [--dburl DBURL]
                                      [--num NUM] [--profile] [--dump]
                                      [--runsnake] [--echo]

                                      {bulk_inserts,large_resultsets,single_inserts}

positional arguments:
  {bulk_inserts,large_resultsets,single_inserts}
                        suite to run

optional arguments:
  -h, --help            show this help message and exit
  --test TEST           run specific test name
  --dburl DBURL         database URL, default sqlite:///profile.db
  --num NUM             Number of iterations/items/etc for tests; default is 0
                        module-specific
  --profile             run profiling and dump call counts
  --dump                dump full call profile (implies --profile)
  --runsnake            invoke runsnakerun (implies --profile)
  --echo                Echo SQL output

示例运行如下所示：

$ python -m examples.performance bulk_inserts

或者选择：

$ python -m examples.performance bulk_inserts \
    --dburl mysql+mysqldb://scott:tiger@localhost/test \
    --profile --num 1000

也可以看看

How can I profile a SQLAlchemy powered application?

档案清单

文件清单：

  • single_inserts.py - 在这一系列测试中，我们正在研究一种在独立事务中插入一行的方法，然后返回到基本上“关闭”状态。这将类似于启动数据库连接，插入行，提交和关闭的API调用。
  • short_selects.py - 这一系列测试说明了通过主键选择单个记录的不同方式
  • bulk_updates.py - 这一系列测试说明了批量更新大量行的不同方法。
  • bulk_inserts.py - 这一系列测试说明了批量插入大量行的不同方法。
  • large_resultsets.py - 在这一系列测试中，我们正在考虑加载大量非常小且简单的行。
  • __ main __。py - 允许将示例/性能包作为脚本运行。

用时间运行所有测试

这是运行的默认形式：

$ python -m examples.performance single_inserts
Tests to run: test_orm_commit, test_bulk_save,
              test_bulk_insert_dictionaries, test_core,
              test_core_query_caching, test_dbapi_raw_w_connect,
              test_dbapi_raw_w_pool

test_orm_commit : Individual INSERT/COMMIT pairs via the
    ORM (10000 iterations); total time 13.690218 sec
test_bulk_save : Individual INSERT/COMMIT pairs using
    the "bulk" API  (10000 iterations); total time 11.290371 sec
test_bulk_insert_dictionaries : Individual INSERT/COMMIT pairs using
    the "bulk" API with dictionaries (10000 iterations);
    total time 10.814626 sec
test_core : Individual INSERT/COMMIT pairs using Core.
    (10000 iterations); total time 9.665620 sec
test_core_query_caching : Individual INSERT/COMMIT pairs using Core
    with query caching (10000 iterations); total time 9.209010 sec
test_dbapi_raw_w_connect : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection each time (10000 iterations); total time 9.551103 sec
test_dbapi_raw_w_pool : Individual INSERT/COMMIT pairs w/ DBAPI +
    connection pool (10000 iterations); total time 8.001813 sec

为单个测试转储配置文件

Python配置文件输出可以转储所有测试，或更常见的单个测试：

$ python -m examples.performance single_inserts --test test_core --num 1000 --dump
Tests to run: test_core
test_core : Individual INSERT/COMMIT pairs using Core. (1000 iterations); total fn calls 186109
         186109 function calls (186102 primitive calls) in 1.089 seconds

   Ordered by: internal time, call count

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    0.634    0.001    0.634    0.001 {method 'commit' of 'sqlite3.Connection' objects}
     1000    0.154    0.000    0.154    0.000 {method 'execute' of 'sqlite3.Cursor' objects}
     1000    0.021    0.000    0.074    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/sql/compiler.py:1950(_get_colparams)
     1000    0.015    0.000    0.034    0.000 /Users/classic/dev/sqlalchemy/lib/sqlalchemy/engine/default.py:503(_init_compiled)
        1    0.012    0.012    1.091    1.091 examples/performance/single_inserts.py:79(test_core)

    ...

使用RunSnake

该选项需要安装RunSnake命令行工具：

$ python -m examples.performance single_inserts --test test_core --num 1000 --runsnake

将显示图形RunSnake输出。

编写自己的套房

profiler套件系统是可扩展的，可以应用于您自己的一套测试。这是一个有价值的技术，用于决定一些性能关键的例程的正确方法。例如，如果我们想分析几种加载之间的差异，我们可以创建一个文件test_loads.py，其中包含以下内容：

from examples.performance import Profiler
from sqlalchemy import Integer, Column, create_engine, ForeignKey
from sqlalchemy.orm import relationship, joinedload, subqueryload, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = None
session = None


class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")


class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))


# Init with name of file, default number of items
Profiler.init("test_loads", 1000)


@Profiler.setup_once
def setup_once(dburl, echo, num):
    "setup once.  create an engine, insert fixture data"
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    sess = Session(engine)
    sess.add_all([
        Parent(children=[Child() for j in range(100)])
        for i in range(num)
    ])
    sess.commit()


@Profiler.setup
def setup(dburl, echo, num):
    "setup per test.  create a new Session."
    global session
    session = Session(engine)
    # pre-connect so this part isn't profiled (if we choose)
    session.connection()


@Profiler.profile
def test_lazyload(n):
    "load everything, no eager loading."

    for parent in session.query(Parent):
        parent.children


@Profiler.profile
def test_joinedload(n):
    "load everything, joined eager loading."

    for parent in session.query(Parent).options(joinedload("children")):
        parent.children


@Profiler.profile
def test_subqueryload(n):
    "load everything, subquery eager loading."

    for parent in session.query(Parent).options(subqueryload("children")):
        parent.children

if __name__ == '__main__':
    Profiler.main()

我们可以直接运行我们的新脚本：

$ python test_loads.py  --dburl postgresql+psycopg2://scott:tiger@localhost/test
Running setup once...
Tests to run: test_lazyload, test_joinedload, test_subqueryload
test_lazyload : load everything, no eager loading. (1000 iterations); total time 11.971159 sec
test_joinedload : load everything, joined eager loading. (1000 iterations); total time 2.754592 sec
test_subqueryload : load everything, subquery eager loading. (1000 iterations); total time 2.977696 sec

以及RunSnake输出的个人测试：

$ python test_loads.py  --num 100 --runsnake --test test_joinedload

关系连接条件

各种orm.relationship()配置的示例，它们利用primaryjoin参数来组合特殊类型的连接条件。

文件清单：

  • cast.py - 说明连接两列的relationship()，其中这些列的类型不同，并且必须在SQL端使用CAST以便匹配他们。
  • threeway.py - Illustrate a “three way join” - where a primary table joins to a remote table via an association table, but then the primary table also needs to refer to some columns in the remote table directly.

XML持久

举例说明了在关系数据库中用ElementTree表示的持久化和查询XML文档的三种策略。这些技术不直接对ElementTree对象应用任何映射，因此与本地cElementTree以及lxml兼容，并且可以适应任何类型的DOM表示系统。也显示了沿类似xpath的字符串查询。

例如。：

# parse an XML file and persist in the database
doc = ElementTree.parse("test.xml")
session.add(Document(file, doc))
session.commit()

# locate documents with a certain path/attribute structure
for document in find_document('/somefile/header/field2[@attr=foo]'):
    # dump the XML
    print document

文件清单：

  • pickle.py - illustrates a quick and dirty way to persist an XML document expressed using ElementTree and pickle.
  • adjacency_list.py - 说明了使用ElementTree保存表示的XML文档的明确方式。
  • optimized_al.py - 使用与adjacency_list.py相同的策略，但将每个DOM行与其拥有的文档行相关联，以便可以使用DOM节点的完整文档加载O（1）查询 - 构建“层次结构”是在非递归方式加载之后执行的，并且效率更高。

版本控制对象

使用历史记录表进行版本控制

说明为实体创建版本表的扩展，并为每个更改存储记录。给定的扩展生成一个匿名的“历史”类，它表示目标对象的历史版本。

用法可以通过单元测试模块test_versioning.py来说明，它可以通过鼻子运行：

cd examples/versioning
nosetests -v

示例用法的片段，使用声明式：

from history_meta import Versioned, versioned_session

Base = declarative_base()

class SomeClass(Versioned, Base):
    __tablename__ = 'sometable'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))

    def __eq__(self, other):
        assert type(other) is SomeClass and other.id == self.id

Session = sessionmaker(bind=engine)
versioned_session(Session)

sess = Session()
sc = SomeClass(name='sc1')
sess.add(sc)
sess.commit()

sc.name = 'sc1modified'
sess.commit()

assert sc.version == 2

SomeClassHistory = SomeClass.__history_mapper__.class_

assert sess.query(SomeClassHistory).\
            filter(SomeClassHistory.version == 1).\
            all() \
            == [SomeClassHistory(version=1, name='sc1')]

Versioned mixin设计用于声明式。要使用经典映射器的扩展，可以应用_history_mapper函数：

from history_meta import _history_mapper

m = mapper(SomeClass, sometable)
_history_mapper(m)

SomeHistoryClass = SomeClass.__history_mapper__.class_

文件清单：

使用时间行的版本控制

说明通过为每个更改存储新行来对数据进行版本升级的扩展；也就是说，通常UPDATE变成INSERT。

文件清单：

  • versioned_rows.py - 说明拦截对象更改的方法，将单个行上的UPDATE语句转换为INSERT语句，以便使用新数据插入新行，使旧行保持不变。
  • versioned_map.py - versioned_rows示例的变体。在这里，我们存储一个键/值对的字典，以“垂直”方式存储k / v，每个键获得一行。该值被分成两个独立的数据类型，字符串和整数 - 数据类型存储的范围可以根据个人需求进行调整。

垂直属性映射

说明“垂直表”映射。

“垂直表”是指将对象的各个属性作为不同的行存储在表中的技术。“垂直表”技术用于持久化可能具有各种属性的对象，代价是简单的查询控制和简洁。它通常存在于内容/文档管理系统中，以便灵活地表示用户创建的结构。

给出了两种不同的方法。在第二行中，每行都引用一个“数据类型”，其中包含有关存储在属性中的信息类型的信息，例如整数，字符串或日期。

例：

shrew = Animal(u'shrew')
shrew[u'cuteness'] = 5
shrew[u'weasel-like'] = False
shrew[u'poisonous'] = True

session.add(shrew)
session.flush()

q = (session.query(Animal).
     filter(Animal.facts.any(
       and_(AnimalFact.key == u'weasel-like',
            AnimalFact.value == True))))
print 'weasel-like animals', q.all()

文件清单：

继承映射食谱

基本继承映射

如datamapping_inheritance中所述的单表，连接表和混凝土表继承的工作示例。

文件清单：

  • concrete.py - Concrete（table-per-class）继承的例子。
  • single.py - 单表继承示例。
  • joined.py - 联合表（继承的子类）继承示例。

特殊的API

属性工具

举例说明对SQLAlchemy属性管理系统的修改。

文件清单：

水平分片

使用SQLAlchemy Sharding API的基本示例。分片是指跨多个数据库水平缩放数据。

“分片”映射的基本组件是：

  • 多个数据库，每个数据库分配一个'分片ID'
  • 给定要保存的实例的函数，它可以返回单个分片ID；这被称为“shard_chooser”
  • 一个可以返回适用于特定实例标识符的分片ID列表的函数；这被称为“id_chooser”。如果它返回所有分片ID，则将搜索所有分片。
  • 给定一个特定的查询（“query_chooser”），该函数可以返回一个分片ID列表来尝试。如果它返回所有分片ID，则将查询所有分片并将结果连接在一起。

在这个例子中，四个sqlite数据库将以每个数据库为基础存储关于天气数据的信息。我们提供了示例shard_chooser，id_chooser和query_chooser函数。query_chooser说明了对SQL表达式元素的检查，以试图确定被请求的单个分片。

通用分片例程的构建是在多个数据库之间组织实例的问题的一个雄心勃勃的方法。对于一个更为通俗易懂的替代方法，“独立实体”方法是一种以明确的方式将对象分配给不同表（以及潜在的数据库节点）的简单方法 - 在维基上的EntityName中进行了描述。

文件清单：

扩展ORM

Dogpile缓存

演示如何在Query对象中嵌入dogpile.cache功能，以允许完全缓存控制以及从长期缓存中拉取“延迟加载”属性的功能。

在版本0.8中更改：该示例已更新为使用dogpile.cache，将Beaker替换为正在使用的缓存库。

在这个演示中，说明了以下技术：

  • 使用Query的自定义子类
  • 避免查询从自定义缓存源而不是数据库中抽取的基本技术。
  • 使用dogpile.cache进行基本缓存，使用允许全局控制一组固定配置的“区域”。
  • 使用自定义的MapperOption对象来配置查询上的选项，包括在发生延迟加载时调用对象图中深层选项的功能。

例如。：

# query for Person objects, specifying cache
q = Session.query(Person).options(FromCache("default"))

# specify that each Person's "addresses" collection comes from
# cache too
q = q.options(RelationshipCache(Person.addresses, "default"))

# query
print q.all()

要运行，必须安装SQLAlchemy和dogpile.cache，或者在当前的PYTHONPATH上运行。演示会为数据文件创建本地目录，插入初始数据并运行。第二次运行该演示将利用已经存在的缓存文件，并且恰好一个针对两个表的SQL语句将被发出 - 然而，显示的结果将利用从缓存中拉出的数十个lazyload。

演示脚本自身，按照复杂性顺序，以Python模块的形式运行，以便相对导入工作：

python -m examples.dogpile_caching.helloworld

python -m examples.dogpile_caching.relationship_caching

python -m examples.dogpile_caching.advanced

python -m examples.dogpile_caching.local_session_caching

文件清单：

  • environment.py - 根据需要建立数据/缓存文件路径和配置，引导夹具数据。
  • caching_query.py - 表示允许在SQLAlchemy中使用Dogpile缓存的函数和类。引入一个名为FromCache的查询选项。
  • model.py - 数据模型，表示具有多个Address对象的Person，每个对象都有PostalCode，City，Country。
  • fixture_data.py - 安装一些示例数据。在这里，我们有几个美国/加拿大城市的邮政编码。然后，安装100个人记录，每个记录都随机选择一个邮政编码。
  • helloworld.py - 说明如何加载一些数据并缓存结果。
  • relationship_caching.py - 说明如何在关系端点上添加缓存选项，以便延迟缓存从缓存中加载。
  • advanced.py - 说明结合使用FromCache选项的Query，包括前端加载，缓存失效和集合缓存。
  • local_session_caching.py - 到目前为止的一切？本示例创建一个新的dogpile.cache后端，该后端将数据保存在当前会话本地的字典中。删除()会话并且缓存消失。

PostGIS集成

演示帮助嵌入PostGIS功能的技巧的天真示例。

这个例子最初是为了将其推广到一个全面的PostGIS集成层而开发的。我们很高兴地宣布，这已经成为GeoAlchemy。

该例子说明：

  • 一个允许CREATE / DROP与AddGeometryColumn / DropGeometryColumn一起使用的DDL扩展
  • 几何类型以及一些子类型，它们将结果行值转换为支持GIS的对象，并且还与DDL扩展集成。
  • 一个支持GIS的对象，它存储一个原始几何值并为诸如AsText()等函数提供一个工厂。
  • 一个ORM比较器，它可以覆盖映射对象上的标准列方法以生成GIS操作员。
  • 一个属性事件侦听器，拦截字符串并转换为GeomFromText()。
  • 一个独立的操作员示例。

该实施仅限于公开的，众所周知且易于使用的扩展点。

例如。：

print session.query(Road).filter(Road.road_geom.intersects(r1.road_geom)).all()

文件清单：