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>

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

文件清单：

有向图

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

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>提交食谱版本。

文件清单：

大集合

大集合的例子。

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

文件清单：

物化路径

使用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参数来组合特殊类型的连接条件。

文件清单：

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

文件清单：

版本控制对象

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

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

用法可以通过单元测试模块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中所述的单表，连接表和混凝土表继承的工作示例。

文件清单：

特殊的API

属性工具

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

文件清单：

水平分片

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

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

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

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

文件清单：

扩展ORM

Dogpile缓存

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

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

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

例如。：

# 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

文件清单：

PostGIS集成

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

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

该例子说明：

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

例如。：

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

文件清单：