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

SQLAlchemy 1.1文档

ORM示例

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

额外的SQLAlchemy示例(某些用户贡献)可在wiki上的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.pygeneric_fk.py脚本是2007年博客文章使用SQLAlchemy的多态关联中呈现的现代化版本。

文件清单:

  • generic_fk.py - Illustrates a so-called “generic foreign key”, in a similar fashion to that of popular frameworks such as Django, ROR, etc. 这种方法绕过了标准的参照完整性实践,因为“外键”列实际上并没有限制引用任何特定的表;相反,应用程序内部逻辑用于确定引用哪个表。
  • table_per_association.py - Illustrates a mixin which provides a generic association via a individually generated association tables for each parent class. 关联的对象本身被保存在所有父母共享的单个表中。
  • 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 - Illustrates a generic association which persists association objects within individual tables, each one generated to persist those objects on behalf of a particular parent class.

大集合

大集合的例子。

说明当相关对象列表非常大时,与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

档案清单

文件清单:

  • single_inserts.py - In this series of tests, we’re looking at a method that inserts a row within a distinct transaction, and afterwards returns to essentially a “closed” state. 这将类似于启动数据库连接的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输出。

写你自己的套房

分析器套件系统是可扩展的,可以应用于您自己的一套测试。这是一个有价值的技术,用于决定一些性能关键的例程的正确方法。例如,如果我们想分析几种加载之间的差异,我们可以创建一个文件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 - Illustrates an explicit way to persist an XML document expressed using ElementTree.
  • optimize_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 - Illustrates a method to intercept changes on objects, turning an UPDATE statement on a single row into an INSERT statement, so that a new row is inserted with the new data, keeping the old row intact.
  • versioned_map.py - A variant of the versioned_rows example. 在这里,我们存储一个键/值对的字典,以“垂直”的方式存储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 - 具体(table-per-class)继承的例子。
  • single.py - Single-table inheritance example.
  • joined.py - 联合表(继承的子类)继承的例子。

特殊的API

属性工具

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

文件清单:

水平分割

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

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

  • 多个数据库,每个数据库分配一个“分片ID”
  • 一个函数,可以返回一个单一的碎片ID,给定一个实例来保存;这被称为“shard_chooser”
  • 一个可以返回适用于特定实例标识符的分片ID列表的函数;这被称为“id_chooser”。如果它返回所有分片ID,则将搜索所有分片。
  • 给定一个特定的Query(“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 - The datamodel, which represents Person that has multiple Address objects, each with PostalCode, City, Country.
  • fixture_data.py - Installs some sample data. 这里我们有几个美国/加拿大城市的邮政编码。然后,安装100个人员记录,每个记录都随机选择一个邮政编码。
  • helloworld.py - Illustrate how to load some data, and cache the results.
  • relationship_caching.py - 说明如何在关系端点上添加缓存选项,以便延迟从缓存中加载。
  • advanced.py - 说明结合使用FromCache选项的Query,包括前端加载,缓存失效和集合缓存。
  • local_session_caching.py - 到目前为止的一切?这个例子创建了一个新的dogpile.cache后端,它将数据保存在当前会话本地的字典中。删除()会话和缓存不见了。

PostGIS集成

一个天真的例子,说明帮助嵌入PostGIS功能的技术。

这个例子最初是为了将其推广到一个全面的PostGIS集成层而开发的。We are pleased to announce that this has come to fruition as GeoAlchemy.

这个例子说明:

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

实施仅限于公共的,众所周知的和简单易用的扩展点。

例如。:

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

文件清单: