zope.sqlalchemy 3.1

简介

本包的目的是统一现有的多种将 SQLAlchemy 与 Zope 事务管理集成的包。因此，它只提供数据管理器，并不试图定义一种配置引擎的 zopeish 方式。

对于 WSGI 应用，可以使用 repoze.tm2（Turbogears 2 和其他系统使用）实现 Zope 风格的自动事务管理。

该包还用于 pyramid_tm（Pyramid 的附加组件）Web 框架。

为了理解此包以及本 README，您需要了解 SQLAlchemy 和 Zope 事务管理器。

运行测试

此包作为 buildout 分发。使用您想要的 python 运行

$ python bootstrap.py $ ./bin/buildout

这将下载依赖包并设置测试脚本，可以使用以下命令运行

$ ./bin/test

或使用标准的 setuptools 测试命令

$ ./bin/py setup.py test

要启用与您自己的数据库的测试，请设置 TEST_DSN 环境变量为您的 sqlalchemy 数据库 dsn。可以通过设置 TEST_TWOPHASE 变量为一个非空字符串来测试两阶段提交行为。例如

$ TEST_DSN=postgres://test:test@localhost/test TEST_TWOPHASE=True bin/test

简要用法

使用会话工厂类上的 register() 函数在 Zope 事务与 SQLAlchemy 事件系统之间进行集成。

from zope.sqlalchemy import register
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

engine = sqlalchemy.create_engine("postgresql://scott:tiger@localhost/test")

DBSession = scoped_session(sessionmaker(bind=engine))
register(DBSession)

现在，实例化会话的提交和回滚将与 Zope 事务集成。

import transaction
from sqlalchemy.sql import text

session = DBSession()

result = session.execute(text("DELETE FROM objects WHERE id=:id"), {"id": 2})
row = result.fetchone()

transaction.commit()

完整示例

以下示例直接来自 SQLAlchemy 声明式文档。首先，必要的导入。

>>> from sqlalchemy import *
>>> from sqlalchemy.orm import declarative_base, scoped_session, sessionmaker, relationship
>>> from sqlalchemy.sql import text
>>> from zope.sqlalchemy import register
>>> import transaction

现在定义映射类。

>>> Base = declarative_base()
>>> class User(Base):
...     __tablename__ = 'test_users'
...     id = Column('id', Integer, primary_key=True)
...     name = Column('name', String(50))
...     addresses = relationship("Address", backref="user")
>>> class Address(Base):
...     __tablename__ = 'test_addresses'
...     id = Column('id', Integer, primary_key=True)
...     email = Column('email', String(50))
...     user_id = Column('user_id', Integer, ForeignKey('test_users.id'))

创建一个引擎并设置表。请注意，为了使此示例工作，需要 sqlite/pysqlite 的最新版本。3.4.0 似乎足够。

>>> engine = create_engine(TEST_DSN)
>>> Base.metadata.create_all(engine)

现在创建会话本身。由于 Zope 是一个线程化 Web 服务器，我们必须使用作用域会话。通过使用 register 将 Zope 和 SQLAlchemy 会话绑定在一起。

>>> Session = scoped_session(sessionmaker(bind=engine,
... twophase=TEST_TWOPHASE))

调用作用域会话工厂以检索会话。您可以在事务中调用此操作任意次数，您将始终检索到相同的会话。目前数据库中没有用户。

>>> session = Session()
>>> register(session)
<zope.sqlalchemy.datamanager.ZopeTransactionEvents object at ...>
>>> session.query(User).all()
[]

现在我们可以创建一个新用户并使用 Zope 的事务机制提交更改，就像 Zope 的发布者一样。

>>> session.add(User(id=1, name='bob'))
>>> transaction.commit()

引擎级别的连接不在事务集成范围内。

>>> engine.connect().execute(text('SELECT * FROM test_users')).fetchall()
[(1, ...'bob')]

新事务需要新的会话。让我们添加一个地址。

>>> session = Session()
>>> bob = session.query(User).all()[0]
>>> str(bob.name)
'bob'
>>> bob.addresses
[]
>>> bob.addresses.append(Address(id=1, email='[email protected]'))
>>> transaction.commit()
>>> session = Session()
>>> bob = session.query(User).all()[0]
>>> bob.addresses
[<Address object at ...>]
>>> str(bob.addresses[0].email)
'[email protected]'
>>> bob.addresses[0].email = 'wrong@wrong'

要回滚事务，请使用 transaction.abort()。

>>> transaction.abort()
>>> session = Session()
>>> bob = session.query(User).all()[0]
>>> str(bob.addresses[0].email)
'[email protected]'
>>> transaction.abort()

默认情况下，zope.sqlalchemy 在会话首次使用时将其置于“活动”状态。ORM 写操作会自动将会话移动到“更改”状态。这避免了不必要的数据库提交。有时需要通过 SQL 直接与数据库交互。无法猜测此类操作是读取还是写入。因此，在手动 SQL 语句写入数据库时，我们必须手动将会话标记为更改。

>>> session = Session()
>>> conn = session.connection()
>>> users = Base.metadata.tables['test_users']
>>> conn.execute(users.update().where(users.c.name=='bob'), {'name': 'ben'})
<sqlalchemy.engine... object at ...>
>>> from zope.sqlalchemy import mark_changed
>>> mark_changed(session)
>>> transaction.commit()
>>> session = Session()
>>> str(session.query(User).all()[0].name)
'ben'
>>> transaction.abort()

如果这成为问题，您可以通过注册事件并告诉它们将会话初始置于“更改”状态来解决这个问题。

>>> Session.remove()
>>> register(Session, 'changed')
<zope.sqlalchemy.datamanager.ZopeTransactionEvents object at ...>
>>> session = Session()
>>> conn = session.connection()
>>> conn.execute(users.update().where(users.c.name=='ben'), {'name': 'bob'})
<sqlalchemy.engine... object at ...>
>>> transaction.commit()
>>> session = Session()
>>> str(session.query(User).all()[0].name)
'bob'
>>> transaction.abort()

mark_changed 函数接受一个 kwarg，用于 keep_session，默认为 False，并且不了解已注册的扩展 keep_session 配置。

如果您打算将 keep_session 设置为 True，您可以明确指定它

>>> from zope.sqlalchemy import mark_changed
>>> mark_changed(session, keep_session=True)
>>> transaction.commit()

您还可以使用配置的扩展来保留此参数

>>> sessionExtension = register(session, keep_session=True)
>>> sessionExtension.mark_changed(session)
>>> transaction.commit()

开发版本

GIT版本

变更

from zope.sqlalchemy import ZopeTransactionExtension

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension(), **options))

from zope.sqlalchemy import register

DBSession = scoped_session(sessionmaker(**options))
register(DBSession)