zc.zodbwsgi 1.2.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)]

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

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

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

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

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

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

生成应用程序具有一个数据库属性（主要用于测试）与创建的数据库。由于是新初始化的，数据库是空的

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

让我们做一个“增量”请求。

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

现在，如果我们查看数据库，我们会看到根对象中现在有数据

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

数据库初始化

我们可以使用initializer选项提供数据库初始化函数。让我们定义一个初始化函数

import transaction

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

并更新我们的paste配置以使用它

[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

现在，当我们使用应用程序时，我们可以看到初始化器的影响

>>> 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数据库，你可能这样做 [1]。你可以通过为transaction_management选项提供“false”的值来抑制事务管理

[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
transaction_management = false

抑制请求重试

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

>>> import ZODB.POSException
>>> app = paste.deploy.loadapp('config:'+os.path.abspath('paste.ini'))
>>> testapp = webtest.TestApp(app)
>>> try: testapp.get('/conflict')
... except ZODB.POSException.ConflictError: pass
... else: print 'oops'
Conflict!
Conflict!
Conflict!
Conflict!

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

我们可以通过为retry选项提供“0”的值来抑制此行为

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

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

retry = 0

现在，如果我们运行应用程序，请求将不会被重试

>>> app = paste.deploy.loadapp('config:'+os.path.abspath('paste.ini'))
>>> testapp = webtest.TestApp(app)
>>> try: testapp.get('/conflict')
... except ZODB.POSException.ConflictError: pass
... else: print 'oops'
Conflict!

使用非线程感知（非线程局部）事务管理器

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

[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

可以通过thread_transaction_manager键来控制此行为

[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
thread_transaction_manager = false

demostorage_manage_header

为此选项提供一个值将启用允许推送/弹出底层demostorage的钩子。

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

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

key = connection
transaction_key = manager
demostorage_manage_header = X-FOO

如果提供了推送或弹出头，中间件将立即返回响应，而不会将其发送到管道的末端。

>>> testapp.get('/', {}, headers={'X-FOO': 'push'}).body
'Demostorage pushed\n'

>>> testapp.get('/inc')
<200 OK text/html body="{'x': 2}">

>>> testapp.get('/', {}, {'X-FOO': 'pop'}).body
'Demostorage popped\n'

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

如果您有访问中间件对象，可以通过调用push和pop方法（这些方法也会返回数据库）来实现相同的功能。这在您在测试过程中运行服务器并具有Python访问权限时非常有用。

>>> db = app.application.push()

请注意，app是repoze.retry，因此我们必须使用.application来获取wsgi应用。

>>> with db.transaction() as conn:
...     conn.root.x = 41

>>> testapp.get('/inc')
<200 OK text/html body="{'x': 42}">

>>> db = app.application.pop()
>>> with db.transaction() as conn:
...     print conn.root.x
1

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

这也适用于多个数据库。

class demo_app:
    def __init__(self, default):
        pass
    def __call__(self, environ, start_response):
        start_response('200 OK', [('content-type', 'text/html')])
        path = environ['PATH_INFO']
        root_one = environ['connection'].get_connection('one').root()
        root_two = environ['connection'].get_connection('two').root()
        if path == '/inc':
            root_one['x'] = root_one.get('x', 0) + 1
            root_two['y'] = root_two.get('y', 0) + 1
            environ['manager'].get().note('path: %r' % path)

        data = {'one': root_one,
                'two': root_two}

        return [repr(data)]

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

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

key = connection
transaction_key = manager
demostorage_manage_header = X-FOO

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

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

[filter:zodb]
use = egg:zc.zodbwsgi
configuration =
   <zodb one>
     <demostorage>
     </demostorage>
   </zodb>
   <zodb two>
     <filestorage>
       path /tmp/Data.fs
     </filestorage>
   </zodb>

key = connection
transaction_key = manager
demostorage_manage_header = foo

限制连接数

如果您使用的是线程服务器，即每个活动请求都分配一个线程的服务器，您可以通过指定max_connections选项来限制同时的数据库连接数。

（这仅适用于线程服务器，因为它使用了线程信号量。将来可能会添加对其他锁定机制的支持，例如gevent信号量。在此期间，如果您想进行猴子补丁，可以将zc.zodbwsgi.Semaphore替换为其他信号量实现，例如gevent的。）

避免连接和事务管理

通常，由您管理连接和事务非常方便。然而，有时您可能希望自行管理事务。

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

处理偶尔的长运行请求

数据库连接可能相当昂贵的资源，特别是如果它们有大的数据库缓存。因此，当使用大型缓存时，通常限制应用程序线程的数量，以限制使用的连接数。如果您的应用程序是计算密集型的，您通常希望每个进程使用一个应用程序线程，并且在主机机器上每个处理器使用一个进程。

如果您的应用程序本身进行网络请求（例如调用外部服务API），那么它是网络/服务器绑定的而不是计算绑定的，您应该增加应用程序线程的数量，并减小连接缓存的大小来补偿。

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

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

• 将max_connections设置为1。

• 在您的应用程序中执行外部服务调用部分

  • 如果需要，首先提交，然后关闭environ['zodb.connection']。

  • 执行服务调用。

  • 当您需要使用数据库时，自行打开和关闭ZODB连接。

    如果您正在使用ZEO或relstorage，您可能希望为这些调用创建单独的数据库客户端，并使用较小的缓存进行配置。

1.2.0 (2015-03-08)

• 在事务元数据中记录请求摘要

• 添加了依赖项以反映导入。

1.1.0 (2014-04-22)

• 提供了用于测试和服务器测试过程的Python push 和 pop 方法。

1.0.0 (2013-09-15)

• 添加了对偶尔的长运行请求的支持

  • 您可以使用max_connections限制数据库连接数。

  • 您可以接管连接和事务管理以在阻塞时（通常在调用外部服务时）释放连接。

• 添加了使用线程感知事务管理器的选项，并将其设置为默认值。

0.3.0 (2013-03-07)

• 现在使用demostorage钩子会立即返回响应，而不会处理管道的其余部分。这使得使用此功能更加清晰。

0.2.1 (2013-03-06)

• 修复了对已重命名的文件的引用。

0.2.0 (2013-03-06)

• 根据标题添加了管理（push/pop）底层demostorage的钩子。

• 重构过滤器以使用实例属性而不是闭包。

0.1.0 (2010-02-16)

初始发布