z3c.saconfig 1.0

简介

该软件包的目标是提供一个简单但灵活的方法，通过 Zope 组件架构来配置 SQLAlchemy 的作用域会话支持。该软件包基于 zope.sqlalchemy，它提供了 Zope 和 SQLAlchemy 之间的事务集成。

这里概述了两种主要场景

• 每个 Zope 实例一个数据库。

• 在 Zope 实例中的每个站点（或 Grok 应用）一个数据库（因此每个 Zope 实例有多个数据库）。

全局作用域会话（每个 Zope 实例一个数据库）

为 Zope 设置 SQLAlchemy 的最简单方法是使用一个全局线程作用域会话。多个应用程序将共享此会话。使用全局实用工具设置引擎。

我们使用 SQLAlchemy sqlalchemy.ext.declarative 扩展来定义一些表和类

>>> from sqlalchemy import *
>>> from sqlalchemy.orm import declarative_base
>>> from sqlalchemy.orm import relationship

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

到目前为止，这与 zope.sqlalchemy 示例没有区别。我们现在到达第一个区别。我们不是直接创建引擎，而是可以将引擎工厂设置为（全局）实用工具。这个实用工具确保为我们创建并缓存一个引擎。

>>> from z3c.saconfig import EngineFactory
>>> engine_factory = EngineFactory(TEST_DSN)

您可以将通常传递给 sqlalchemy.create_engine 的参数传递给 EngineFactory。

现在，我们使用 zope.component 将引擎工厂注册为全局实用工具。通常您会使用 ZCML 或 Grok 来执行此确认，但我们将在这里手动执行：

>>> from zope import component
>>> from z3c.saconfig.interfaces import IEngineFactory
>>> component.provideUtility(engine_factory, provides=IEngineFactory)

请注意，在全局作用域用例中设置引擎工厂实际上不是必需的。您也可以在创建 GloballyScopedSession 时将其作为全局创建的引擎并传递给 bind。

让我们通过调用工厂来查找引擎，并在我们的测试数据库中创建表

>>> engine = engine_factory()
>>> Base.metadata.create_all(engine)

现在，关于与 zope.sqlalchemy 的第二个区别：会话的设置和使用方式。我们将使用 GloballyScopedSession 实用工具来实现会话创建

>>> from z3c.saconfig import GloballyScopedSession

我们将构造函数传递给 GloballyScopedSession 的参数通常传递给 sqlalchemy.orm.create_session 或 sqlalchemy.orm.sessionmaker

>>> utility = GloballyScopedSession(twophase=TEST_TWOPHASE)

GlobalScopedSession 将使用 IEngineFactory 查找引擎，如果您没有提供自己的 bind 参数。 GlobalScopedSession 还自动设置 autocommit、autoflush 和 extension 参数为与 Zope 集成正确的值，所以通常您不需要提供这些，但如果您需要，您可以提供自己的。

现在我们将它注册为 IScopedSession 实用工具

>>> from z3c.saconfig.interfaces import IScopedSession
>>> component.provideUtility(utility, provides=IScopedSession)

配置已完成。如您所见，这涉及到设置两个工具，IEngineFactory 和 IScopedSession，其中只有后者在全局共享会话用例中是真正必需的。

在注册了 IScopedSession 工具之后，可以从 z3c.saconfig 导入 Session 类。这个 Session 类类似于您使用 SQLAlchemy 的 sessionmaker 产生的类。z3c.saconfig.Session 的目的是您唯一需要的 Session 类，因为所有配置和 Zope 集成都由 z3c.saconfig 自动为您完成，适合您在 Zope 中使用它的上下文。您不再需要使用 sessionmaker 或 scoped_sesion 自己创建 Session 类。

现在我们可以使用 Session 类创建一个会话，该会话将根据我们提供的工具的行为。

>>> from z3c.saconfig import Session
>>> session = Session()

现在事情按照通常的 zope.sqlalchemy 方式进行，这类似于 SQLAlchemy，但您可以使用 Zope 的 transaction 模块。

>>> session.query(User).all()
[]
>>> import transaction
>>> session.add(User(name='bob'))
>>> transaction.commit()

>>> session = Session()
>>> bob = session.query(User).all()[0]
>>> bob.name == 'bob'
True
>>> bob.addresses
[]

事件

当一个新引擎由 EngineFactory 创建时，会触发一个 IEngineCreatedEvent。此事件有一个名为 engine 的属性，其中包含刚刚创建的引擎。

>>> from z3c.saconfig.interfaces import IEngineCreatedEvent
>>> @component.adapter(IEngineCreatedEvent)
... def createdHandler(event):
...     print("created engine")
...     print("args: {0}".format(event.engine_args))
...     print("kw: {0}".format(event.engine_kw))
>>> component.provideHandler(createdHandler)
>>> event_engine_factory = EngineFactory(TEST_DSN1)
>>> engine = event_engine_factory()
created engine
args: ('sqlite:///:memory:',)
kw: {}

让我们再次去掉事件处理器。

>>> sm = component.getSiteManager()
>>> sm.unregisterHandler(None,
...   required=[IEngineCreatedEvent])
True

站点作用域会话（每个站点一个数据库）

在上面的例子中，我们使用工具设置了 SQLAlchemy 和 Zope，但这对我们来说并没有带来很大的好处，除了您可以使用 zope.sqlalchemy.Session 来获取正确的会话。

现在我们将看到如何通过为每个站点注册不同的引擎工厂来设置不同的引擎。

为了实现这一点，我们将使用 SiteScopedSession 而不是 GloballyScopedSession。首先我们需要继承 SiteScopedSession，因为我们需要实现其 siteScopeFunc 方法，该方法应返回每个站点的唯一 ID（例如通过 zope.traversing.api.getPath 获取的路径）。我们需要在这里实现它，因为 z3c.saconfig 将此策略留给应用程序或更高层次的框架。

>>> from z3c.saconfig import SiteScopedSession
>>> class OurSiteScopedSession(SiteScopedSession):
...   def siteScopeFunc(self):
...      return getSite().id # the dummy site has a unique id
>>> utility = OurSiteScopedSession()
>>> component.provideUtility(utility, provides=IScopedSession)

我们想要注册两个引擎工厂，每个工厂在不同的站点上。

>>> engine_factory1 = EngineFactory(TEST_DSN1)
>>> engine_factory2 = EngineFactory(TEST_DSN2)

我们需要在两个新的引擎中设置数据库。

>>> Base.metadata.create_all(engine_factory1())
>>> Base.metadata.create_all(engine_factory2())

现在让我们创建两个站点，每个站点都将连接到另一个引擎。

>>> site1 = DummySite(id=1)
>>> site2 = DummySite(id=2)

我们为每个站点设置本地引擎工厂。

>>> sm1 = site1.getSiteManager()
>>> sm1.registerUtility(engine_factory1, provided=IEngineFactory)
>>> sm2 = site2.getSiteManager()
>>> sm2.registerUtility(engine_factory2, provided=IEngineFactory)

为了防止意外发生，我们将禁用我们的全局引擎工厂。

>>> component.provideUtility(None, provides=IEngineFactory)

当我们设置站点为 site1 时，查找 IEngineFactory 会得到引擎工厂 1。

>>> setSite(site1)
>>> component.getUtility(IEngineFactory) is engine_factory1
True

当我们将其设置为 site2 时，我们会得到引擎工厂 2。

>>> setSite(site2)
>>> component.getUtility(IEngineFactory) is engine_factory2
True

即使我们处于站点中，我们也可以查找我们的全局工具。

>>> component.getUtility(IScopedSession) is utility
True

哇，设置了很多，但实际上这只是一个直接的实用工具设置代码；您应该使用 API 或 Grok 的 grok.local_utility 指令来设置本地实用工具。现在，所有这些都已完成，我们可以为 site1 创建一个会话。

>>> setSite(site1)
>>> session = Session()

数据库仍然是空的。

>>> session.query(User).all()
[]

现在我们将向这个数据库添加一些内容。

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

bob 现在在那里了。

>>> session = Session()
>>> session.query(User).all()[0].name == 'bob'
True

现在我们将切换到 site2。

>>> setSite(site2)

如果我们现在创建一个新的会话，我们应该现在正在使用不同的数据库，这个数据库应该仍然是空的

>>> session = Session()
>>> session.query(User).all()
[]

我们将把fred添加到这个数据库中

>>> session.add(User(name='fred'))
>>> transaction.commit()

现在fred确实在那里

>>> session = Session()
>>> users = session.query(User).all()
>>> len(users)
1
>>> users[0].name == 'fred'
True

并且bob仍然在site1中

>>> setSite(site1)
>>> session = Session()
>>> users = session.query(User).all()
>>> len(users)
1
>>> users[0].name == 'bob'
True

引擎和多线程

>>> engine = None
>>> def setEngine():
...     global engine
...     engine = engine_factory1()

引擎工厂必须生成相同的引擎

>>> setEngine()
>>> engine is engine_factory1()
True

即使你在不同的线程中调用它

>>> import threading
>>> engine = None
>>> t = threading.Thread(target=setEngine)
>>> t.start()
>>> t.join()

>>> engine is engine_factory1()
True

除非它们被重置

>>> engine_factory1.reset()
>>> engine is engine_factory1()
False

即使是在几乎相同时间创建的具有相同参数的引擎工厂也应该生成不同的引擎

>>> EngineFactory(TEST_DSN1)() is EngineFactory(TEST_DSN1)()
False

使用ZCML进行配置

提供了一个配置指令，用于使用ZCML注册数据库引擎工厂。

>>> from io import BytesIO
>>> from zope.configuration import xmlconfig
>>> import z3c.saconfig
>>> xmlconfig.XMLConfig('meta.zcml', z3c.saconfig)()

让我们再次尝试注册目录。

>>> xmlconfig.xmlconfig(BytesIO(b"""
... <configure xmlns="http://namespaces.zope.org/db">
...   <engine name="dummy" url="sqlite:///:memory:" />
... </configure>"""))

>>> component.getUtility(IEngineFactory, name="dummy")
<z3c.saconfig.utility.EngineFactory object at ...>

这次使用设置调用。

>>> xmlconfig.xmlconfig(BytesIO(b"""
... <configure xmlns="http://namespaces.zope.org/db">
...   <engine name="dummy2" url="sqlite:///:memory:"
...           setup="z3c.saconfig.tests.engine_subscriber" />
... </configure>"""))
got: Engine(sqlite:///:memory:)

还可以指定连接池选项

>>> xmlconfig.xmlconfig(BytesIO(b"""
... <configure xmlns="http://namespaces.zope.org/db">
...   <engine name="dummy" url="sqlite:///:memory:"
...       pool_size="1"
...       max_overflow="2"
...       pool_recycle="3"
...       pool_timeout="4"
...       />
... </configure>"""))

>>> engineFactory = component.getUtility(IEngineFactory, name="dummy")
>>> engineFactory._kw == {'echo': None, 'pool_size': 1, 'max_overflow': 2, 'pool_recycle': 3, 'pool_timeout': 4}
True

(有关这些参数如何使用的详细信息，请参阅SQLAlchemy关于连接池的文档。)

提供session指令以注册作用域会话实用工具

>>> xmlconfig.xmlconfig(BytesIO(b"""
... <configure xmlns="http://namespaces.zope.org/db">
...   <session name="dummy" engine="dummy2" />
... </configure>"""))

>>> component.getUtility(IScopedSession, name="dummy")
<z3c.saconfig.utility.GloballyScopedSession object at ...>

>>> from z3c.saconfig import named_scoped_session
>>> factory = component.getUtility(IEngineFactory, name="dummy2")
>>> Session = named_scoped_session('dummy')
>>> Session().bind is factory()
True

1.0 (2023-06-13)

• 支持Python 3.9、3.10、3.11。

• 停止支持Python 2.7、3.5、3.6。

• 更新测试以与SQLAlchemy 2一起运行。(无法保证它们仍然可以在旧版本中运行。)

• 在ZCML engine指令中忽略convert_unicode参数，因为它不再由SQLAlchemy 2支持。

0.16.0 (2020-04-03)

• 增加了对Python 3.7的支持[nazrulworld]

• 增加了对Python 3.8的支持[icemac]

• 增加了对zope.sqlalchemy >= 1.2的支持[cklinger]

• 更新了本地bootstrap.py [cklinger]

• 使用较新的SQLAlchemy进行测试[cklinger]

0.15 (2018-11-30)

• 增加了Python 3.5和3.6的兼容性[nazrulworld]

• 修复：在python3兼容性问题，在zope接口实现上 [nazrulworld]

0.14 (2015-06-29)

• 停止支持sqlalchemy < 0.5 [oggers]

0.13 (2011-07-26)

• 使用zcml操作注册引擎工厂设置

0.12 (2010-09-28)

• EngineCreatedEvent也获得engine_args和engine_kw作为属性，这样事件处理器可以潜在地区分引擎。

0.11 (2010-07-05)

• 向

• 与sqlalchemy >= 0.5兼容(与sqlalchemy > 5之前不兼容)

0.10 (2010-01-18)

• 支持当前ZTK代码

• engine.echo必须默认为None，以便SQLAlchemy可以尊重logging.getLogger(“sqlalchemy.engine”).setLevel(…)

• 默认不启用convert_unicode。此选项通过使String类型列返回Unicode数据而更改标准SQLAlchemy行为。在Unicode不总是被接受的Zope2环境中，这可能会特别痛苦。

• 在zcml引擎语句中添加convert_unicode选项，允许需要convert_unicode的人启用它。

0.9.1 (2009-08-14)

• 包含PyPI上的文档。

• 小的文档调整。

0.9 (2009-08-14)

• 首次公开发布。