pyramid-services 2.2

安装

使用虚拟环境中的pip或easy_install从PyPI安装。

$ $VENV/bin/pip install pyramid_services

或直接从源安装。

$ git clone https://github.com/mmerickel/pyramid_services.git
$ cd pyramid_services
$ $VENV/bin/pip install -e .

设置

通过将其包含到Pyramid应用程序中激活pyramid_services。

config.include('pyramid_services')

这将向您的Configurator添加一些新的指令。

• config.register_service(obj, iface=Interface, context=Interface, name='')

  此方法将为提供的iface、context和name注册一个服务对象。这实际上为应用程序注册了一个单例，因为当查找服务时将始终返回obj。

• config.register_service_factory(factory, iface=Interface, context=Interface, name='')

  此方法将为提供的 iface、context 和 name 注册一个工厂。工厂应该是一个可调用的对象，接受一个 context 和一个 request，并返回一个服务对象。工厂将最多在每个 request/context/name 组合中使用一次。

• config.set_service_registry(registry)

  此方法将允许您设置一个自定义的 wired.ServiceRegistry 实例，它是所有服务的后端注册表。

用法

在通过 Configurator 注册服务后，它们现在可以通过 request.find_service(iface=Interface, context=_marker, name='') 方法在请求生命周期中通过请求对象访问。除非向 find_service 传递了自定义的 context，否则查找将默认使用 request.context。如果在 Pyramid 中进行遍历期间或之前搜索服务时可能没有 request.context，则 context 将默认为 None。

svc = request.find_service(ILoginService)

from pyramid_services import NewServiceContainer

def on_new_container(event):
    container = event.container
    request = event.request
    container.set(request.tm, name='tm')

config.add_subscriber(on_new_container, NewServiceContainer)

示例

让我们从头开始逐步构建我们想在应用程序中使用的登录服务。

基本上，配置接口的所有步骤都是可选的，但在这里它们被展示为最佳实践。

# myapp/interfaces.py

from zope.interface import Interface

class ILoginService(Interface):
  def create_token_for_login(name):
    pass

有了我们的接口，我们现在可以定义一个符合要求的实例。

# myapp/services.py

class DummyLoginService(object):
  def create_token_for_login(self, name):
    return 'u:{0}'.format(name)

让我们将其连接到我们的应用程序。

# myapp/main.py

from pyramid.config import Configurator

from myapp.services import DummyLoginService

def main(global_config, **settings):
  config = Configurator()
  config.include('pyramid_services')

  config.register_service(DummyLoginService(), ILoginService)

  config.add_route('home', '/')
  config.scan('.views')
  return config.make_wsgi_app()

最后，让我们创建一个利用服务的视图。

# myapp/views.py

@view_config(route_name='home', renderer='json')
def home_view(request):
  name = request.params.get('name', 'bob')

  login_svc = request.find_service(ILoginService)
  token = login_svc.create_token_for_login(name)

  return {'access_token': token}

如果您启动此应用程序，您会发现您可以通过访问主页并获取自定义令牌！

这很酷，但更好的是，我们可以完全不用更改视图就替换新的服务。让我们定义一个新的 PersistentLoginService，它从数据库中获取令牌。我们将需要设置一些数据库处理，但在视图中没有改变。

# myapp/services.py

from uuid import uuid4

from myapp.model import AccessToken

class PersistentLoginService(object):
  def __init__(self, dbsession):
    self.dbsession = dbsession

  def create_token_for_login(self, name):
    token = AccessToken(key=uuid4(), user=name)
    self.dbsession.add(token)
    return token.key

以下是一些使用优秀的 SQLAlchemy ORM 配置模型的样板代码。

# myapp/model.py

from sqlalchemy import engine_from_config
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy.types import Text

Base = declarative_base()

def init_model(settings):
  engine = engine_from_config(settings)
  dbmaker = sessionmaker()
  dbmaker.configure(bind=engine)
  return dbmaker

class AccessToken(Base):
  __tablename__ = 'access_token'

  key = Column(Text, primary_key=True)
  user = Column(Text, nullable=False)

现在我们将更新应用程序以使用新的 PersistentLoginService。然而，我们可能还有其他服务，为每个请求中的每个服务创建一个新的数据库连接是愚蠢的。因此，我们还将添加一个封装数据库连接的服务。使用这种技术，我们可以在服务层中连接服务。

# myapp/main.py

from pyramid.config import Configurator
import transaction
import zope.sqlalchemy

from myapp.model import init_model
from myapp.services import PersistentLoginService

def main(global_config, **settings):
  config = Configurator()
  config.include('pyramid_services')
  config.include('pyramid_tm')

  dbmaker = init_model(settings)

  def dbsession_factory(context, request):
    dbsession = dbmaker()
    # register the session with pyramid_tm for managing transactions
    zope.sqlalchemy.register(dbsession, transaction_manager=request.tm)
    return dbsession

  config.register_service_factory(dbsession_factory, name='db')

  def login_factory(context, request):
    dbsession = request.find_service(name='db')
    svc = PersistentLoginService(dbsession)
    return svc

  config.register_service_factory(login_factory, ILoginService)

  config.add_route('home', '/')
  config.scan('.views')
  return config.make_wsgi_app()

最终，主页视图将保持不变。

# myapp/views.py

@view_config(route_name='home', renderer='json')
def home_view(request):
  name = request.params.get('name', 'bob')

  login_svc = request.find_service(ILoginService)
  token = login_svc.create_token_for_login(name)

  return {'access_token': token}

希望这种模式是清晰的。它比大多数基本的 Pyramid 教程具有几个优点。

• 模型完全从视图中抽象出来，使得两者都易于单独测试。

• 服务层可以独立于视图进行开发，允许为模板和前端逻辑创建模拟实现。稍后，真实的服务层可以替换进来，构建后端功能。

• 大多数服务可以以不依赖于 Pyramid 或特定请求对象的方式实现。

• 根据上下文，可能会返回不同的服务，例如遍历的结果或其他应用定义的判别器。

测试示例

如果你正在编写使用 pyramid_services 的应用程序，你可能需要进行一些集成测试，以验证你的应用程序是否成功调用了 register_service 或 register_service_factory。使用 Pyramid 的 testing 模块创建一个 Configurator，并在调用 config.include('pyramid_services') 之后，你可以使用 find_service_factory 来获取已注册服务的相关信息。

以下是一个示例测试，用于验证 dbsession_factory 是否已正确注册。这假设你有一个包含 includeme() 函数的 myapp.services 包。

# myapp/tests/test_integration.py

from myapp.services import dbsession_factory, login_factory, ILoginService

class TestIntegration_services(unittest.TestCase):
  def setUp(self):
    self.config = pyramid.testing.setUp()
    self.config.include('pyramid_services')
    self.config.include('myapp.services')

  def tearDown(self):
    pyramid.testing.tearDown()

  def test_db_maker(self):
    result = self.config.find_service_factory(name='db')
    self.assertEqual(result, dbsession_factory)

  def test_login_factory(self):
    result = self.config.find_service_factory(ILoginService)
    self.assertEqual(result, login_factory)

2.2 (2019-04-22)

• 使用 wired >= 0.2 依赖项来使用新的 wired.ServiceContainer.register_singleton API。

2.1 (2018-08-03)

• 在调用 request.find_service 之前创建服务容器时，将发出 NewServiceContainer 事件。

2.0 (2018-08-03)

• 停止支持 Python 2.7。

• 在底层用 https://wired.readthedocs.io 替换服务查找。

• 修复了使用自定义上下文进行服务查找的问题，使得上下文可以传递给嵌套的服务查找。

1.1 (2017-05-31)

特性

• 如果 iface 是一个类，那么它必须是最初注册的确切类。由于内部限制，子类目前不被识别为实现相同的接口。

1.0 (2017-05-11)

0.4 (2016-02-03)

向后不兼容性

• 停止支持 Python 3.2。

• 使用原始服务上下文接口作为缓存键，而不是当前上下文。这意味着对于满足原始接口的任何上下文，服务都只会正确创建一次。

  以前，如果你在同一个请求中从两个不同的上下文请求相同的服务，你会收到两个服务对象，而不是原始服务的缓存版本，假设服务已注册以满足这两个上下文。请参阅 https://github.com/mmerickel/pyramid_services/pull/12

0.3 (2015-12-13)

• 在遍历期间或之前使用 request.find_service 时，request.context 无效。在这些情况下，context 参数将默认为 None，而不是引发异常。请参阅 https://github.com/mmerickel/pyramid_services/pull/8

• 添加 config.find_service_factory 和 request.find_service_factory。请参阅 https://github.com/mmerickel/pyramid_services/pull/4

0.2 (2015-03-13)

• 将 find_service(..., context=None) 中的 context 参数更改为 None。以前，如果 context 为 None，则将回退到使用 request.context。现在，只有在未指定 context 参数时，find_service 才会回退到 request.context。请参阅 https://github.com/mmerickel/pyramid_services/pull/3

• 支持服务的 introspectable，以便它们在 pyramid_debugtoolbar 以及其他地方显示。请参阅 https://github.com/mmerickel/pyramid_services/pull/2

0.1.1 (2015-02-17)

• 支持 request.find_service、config.register_service 和 config.register_service_factory。

• 初始提交。