zc.zodbwsgi 1.2.0

用于管理zodb数据库连接的wsgi中间件

zodbwsgi提供了用于管理到zodb的连接的中间件。 数据库。它将几个特性组合到一个中间件中 组件：

• 数据库配置
• 数据库初始化
• 连接管理
• 可选交易管理
• 冲突错误时的可选请求重试（使用repoze.retry）
• 可选地限制同时数据库连接的数量
• 应用程序可以接管 以个案为基础，例如支持 长时间运行的请求。

它设计用于粘贴部署，并提供 "filter_app_factory"入口点，命名为"main"。

提供了许多配置选项。选项值是 弦乐.

< DL>

配置

一个必需的zconfig格式的zodb数据库配置

如果定义了多个数据库，它们将定义 多数据库。将连接到第一个定义的数据库。

初始值设定项

表单的可选数据库初始化功能 模块：表达式 DK>键< /dt >

数据库连接的wsgi环境密钥的可选名称

默认为"zodb.connection"。

交易管理

表示 中间件管理事务。

默认情况下启用事务管理。

交易密钥

事务管理器的wsgi环境密钥的可选名称

默认为"transaction.manager"。钥匙只会是 如果启用了事务管理，则显示。

线程事务管理器

指示 中间件将使用线程感知事务管理器（例如， thread.transactionmanager）。

使用线程感知事务管理器很方便 使用始终在同一线程中请求的服务器，例如 服务器使用线程池，或者为每个线程创建线程 请求，

如果您使用的服务器（如gevent）处理多个 同一线程或服务器中可能处理相同的请求 在不同线程中请求，则应将此选项设置为 错误.

默认为真。

重试

可选的重试次数

默认值为"3"，表示请求将重试到 3次。使用"0"禁用重试。

注意，当retry不是"0"时，请求主体将被缓冲。

演示存储管理标题

控制筛选器是否支持推/弹出的可选项 支持底层的demostorage。

如果提供了一个值，它将在请求中检查该头。如果发现 它的值是"push"或"pop"，它将执行相关操作。这个 中间件将返回一个响应，指示在没有_ 正在处理其余管道。

还要注意，只有在底层存储是demostorage时，此功能才起作用。

最大连接数

同时连接的最大数目。

内容

• 用于管理zodb数据库连接的wsgi中间件
• 基本用法
• 数据库初始化
• 禁用事务管理
• 禁止请求重试
• 使用非线程感知（非线程本地）事务管理器
• demostorage\u manage\u头
• 限制连接数
• 转义连接和事务管理
• 处理偶尔的长时间运行的请求

更改

• 1.2.0（2015-03-08）
• 1.1.0（2014-04-22）
• 1.0.0（2013-09-15）
• 0.3.0（2013-03-07）
• 0.2.1（2013-03-06）
• 0.2.0（2013-03-06）
• 0.1.0（2010-02-16）

基本用法

让我们看一些例子。

首先，我们定义一个演示"应用程序"，可以传递给 工厂：

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

现在，我们将使用粘贴部署定义我们的应用程序工厂 配置：

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

在这里，出于演示目的，我们使用了内存中的演示存储器。

现在，我们将使用paste创建一个应用程序：

< Buff行情>

>>> import paste.deploy, os
>>> app = paste.deploy.loadapp('config:'+os.path.abspath('paste.ini'))

生成的应用程序具有数据库属性（主要用于 使用创建的数据库。 新初始化时，数据库为空：

< Buff行情>

>>> conn = app.database.open()
>>> conn.root()
{}

让我们做一个"增量"请求。

< Buff行情>

>>> import webtest
>>> testapp = webtest.TestApp(app)
>>> testapp.get('/inc')
<200 OK text/html body="{'x': 1}">

现在，如果我们查看数据库，就会发现 根对象：

< Buff行情>

>>> conn.sync()
>>> conn.root()
{'x': 1}

数据库初始化

我们可以使用初始化器提供数据库初始化函数 选择权。让我们定义一个初始化函数：

import transaction

def initialize_demo_db(db):
    conn = db.open()
    conn.root()['x'] = 100
    transaction.commit()
    conn.close()

并更新粘贴配置以使用它：

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

initializer = zc.zodbwsgi.tests:initialize_demo_db

现在，当我们使用这个应用程序时，我们可以看到 初始值设定项：

< Buff行情>

>>> app = paste.deploy.loadapp('config:'+os.path.abspath('paste.ini'))
>>> testapp = webtest.TestApp(app)
>>> testapp.get('/inc')
<200 OK text/html body="{'x': 101}">

禁用事务管理

有时，您可能不希望中间件控制事务。 如果应用程序使用多个数据库， 包括非zodb数据库。你可以抑制 通过为 交易管理选项：

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

0

禁止请求重试

默认情况下，zc.zodbwsgi添加repoze.retry中间件来重试请求 当出现冲突错误时：

< Buff行情>

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

1

在这里，我们可以看到请求被重试了3次。

我们可以通过为重试选项提供值"0"来抑制此现象：

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

2

现在，如果运行应用程序，则不会重试请求：

< Buff行情>

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

3

使用非线程感知（非线程本地）事务管理器

默认情况下，中间件使用线程感知事务管理器：

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

4

这可以通过 线程事务管理器控制

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

5

演示存储管理标题

提供此选项的值将启用允许推送/弹出的挂钩 基本的演示存储。

< Buff行情>

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

6

如果提供了push或pop头，中间件将返回一个响应 立即发送，而不发送到管道末端。

< Buff行情>

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

7

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

8

import transaction, ZODB.POSException
from sys import stdout

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        root = environ['zodb.connection'].root()
        path = environ['PATH_INFO']
        if path == '/inc':
            root['x'] = root.get('x', 0) + 1
            if 'transaction.manager' in environ:
                environ['transaction.manager'].get().note('path: %r' % path)
            else:
                transaction.commit() # We have to commit our own!
        elif path == '/conflict':
            print >>stdout, 'Conflict!'
            raise ZODB.POSException.ConflictError
        elif path == "/tm":
            tm = environ["transaction.manager"]
            return ["thread tm: " + str(tm is transaction.manager)]
        return [repr(root)]

9

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

0

如果您有权访问中间件对象，则可以完成 通过调用push和pop方法，也可以返回 数据库。在测试中运行服务器时，这很有用 处理并具有python访问权限：

< Buff行情>

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

1

请注意，应用程序是一个repoze.retry，因此我们必须使用.application才能获取 wsgi应用程序。

< Buff行情>

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

2

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

3

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

4

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

0

这也适用于多个数据库。

< Buff行情>

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

6

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

7

如果任何数据库的存储不是demostorage，则错误是 返回。

< Buff行情>

[app:main]
paste.app_factory = zc.zodbwsgi.tests:demo_app
filter-with = zodb

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb>
     <demostorage>
     </demostorage>
   </zodb>

8

限制连接数

如果您使用的是线程服务器，则为每个线程分配一个线程 活动请求，您可以限制同时数据库的数量 通过使用 最大连接数指定连接数 选项. < /P>

（这只适用于线程服务器，因为它使用线程 信号量。在未来，支持其他锁定机制，如 作为gevent信号量，可以添加。同时，如果你 倾向于monkey补丁，可以替换 zc.zodbwsgi.semaphore 使用另一种信号量实现，如gevent的实现）

转义连接和事务管理

通常，为您管理连接和事务是 方便。但是，有时您希望接管事务 自己管理。

如果关闭环境['zodb.connection']，则不会关闭 通过 zc.zodbwsgi ，也不会提交或中止 它启动的事务。如果您使用的是 max_connections ，请关闭 环境['zodb.connection'] 将使连接可用于 立即提出其他请求，而不是等待您的请求 完成。

处理偶尔的长时间运行的请求

数据库连接可能是非常昂贵的资源，特别是如果 它们有大型数据库缓存。因此，当使用大型 缓存，通常将应用程序线程的数量限制为 限制使用的连接数。如果你的应用程序是计算的 绑定后，通常希望每个进程使用一个应用程序线程 以及主机上每个处理器的进程。

如果应用程序本身发出网络请求（例如调用 外部服务api），因此它是网络/服务器绑定的，而不是 在计算范围内，应该增加应用程序线程的数量 并减小连接缓存的大小以进行补偿。

如果应用程序主要是计算绑定的，但有时调用 外部服务，您可以采用混合方法：

• 增加应用程序线程的数量。

• 将 最大连接数设置为1。

• 在应用程序中进行外部服务调用的部分：

  • 关闭re">环境['zodb.connection'] ，先提交，如果 必要。

  • 拨打您的服务电话。

  • 当您需要使用 数据库。

    如果您使用的是zeo或relstorage，则可能需要创建 在这些调用中使用的单独数据库客户端，配置为 较小的缓存。