Pyramid#

svcs’s Pyramid integration uses Pyramid’s pyramid.registry.Registry to store its own svcs.Registry (yes, unfortunate name clash) and a tween that attaches a fresh svcs.Container to every request and closes it afterwards.

Initialization#

The most important integration API for Pyramid is svcs.pyramid.init() that takes an pyramid.config.Configurator and optionally the positions where to put its tween using the tween_under and tween_over arguments.

You can use svcs.pyramid.register_factory() and svcs.pyramid.register_value() that work like their svcs.Registry counterparts but take a pyramid.config.Configurator as the first option (or any other object that has a registry: dict field, really).

So you application factory is going to look something like this:

def make_app():
    ...

    with Configurator(settings=settings) as config:
        svcs.pyramid.init(config)
        svcs.pyramid.register_factory(config, Database, db_factory)

        ...

        return config.make_wsgi_app()

Service Acquisition#

You can use svcs.pyramid.svcs_from() to access a request-scoped svcs.Container from a request object:

from svcs.pyramid import svcs_from

def view(request):
    db = svcs_from(request).get(Database)

Or you can use svcs.pyramid.get() to access a service from the current request directly:

import svcs

def view(request):
    db = svcs.pyramid.get(request, Database)

Thread Locals#

Despite being discouraged, you can use Pyramid’s thread locals to access the active container.

So this:

def view(request):
    registry = svcs.pyramid.get_registry()
    container = svcs.pyramid.svcs_from()

is equivalent to this:

def view(request):
    registry = svcs.pyramid.get_registry(request)
    container = svcs.pyramid.svcs_from(request)

Caution

These functions only work from within active Pyramid requests.

Health Checks#

As with services, you have the option to either svcs.pyramid.svcs_from() on the request or go straight for svcs.pyramid.get_pings().

A health endpoint could look like this:

from __future__ import annotations

import json

from pyramid.request import Request
from pyramid.response import Response
from pyramid.view import view_config

import svcs


@view_config(route_name="healthy")
def healthy_view(request: Request) -> Response:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: list[dict[str, str]] = []
    status = 200

    for svc in svcs.pyramid.get_pings(request):
        try:
            svc.ping()
            ok.append(svc.name)
        except Exception as e:
            failing.append({svc.name: repr(e)})
            status = 500

    return Response(
        content_type="application/json",
        status=status,
        body=json.dumps({"ok": ok, "failing": failing}).encode("ascii"),
    )

Testing#

Assuming you have an application factory your_app.make_app()[1] that initializes and configures svcs using svcs.pyramid.init(), you can use the following fixtures to get a WebTest application and its registry for overrides:

from your_app import make_app

import pytest
import svcs
import webtest

@pytest.fixture
def app():
    app = make_app()

    with svcs.pyramid.get_registry(app):
        yield webtest.TestApp(app)


@pytest.fixture
def registry(app):
    return svcs.pyramid.get_registry(app)

Now you can write a test like this:

from sqlalchemy import Engine

def test_broken_database(app, registry):
    boom = Mock(spec_set=Engine)
    boom.execute.side_effect = RuntimeError("Boom!")

    registry.register_value(Engine, boom)  # ← override the database

    resp = app.get("/some-url")

    assert 500 == resp.status_code

Since init() takes a registry keyword argument, you can also go the other way around and pass a (potentially pre-configured) svcs.Registry into your application factory.

Cleanup#

You can use svcs.pyramid.close_registry() to close the registry that is attached to the pyramid.registry.Registry of the config or app object that you pass as the only parameter.

API Reference#

Application Life Cycle#

svcs.pyramid.init(config, *, registry=None, tween_under=None, tween_over=None)[source]#

Configure config to work with svcs.

svcs uses a tween to manage the life cycle of the container. You can affect its position by passing tween_under and tween_over.

Parameters:
svcs.pyramid.close_registry(rh)[source]#

Close the registry on rh, if present.

Ideal for atexit.register() handlers.

Parameters:

rh (PyramidRegistryHaver) – An object that carries a pyramid.registry.Registry.

svcs.pyramid.svcs_from(request=None)[source]#

Get the current container either from request or from thread locals.

Parameters:

request (Request | None) – If None, thread locals are used.

svcs.pyramid.get_registry(rh=None)[source]#

Get the registry from rh or thread locals.

Parameters:

rh (PyramidRegistryHaver | None) – If None, thread locals are used.

class svcs.pyramid.PyramidRegistryHaver[source]#

An object with a pyramid.registry.Registry as a registry attribute. For example a Configurator or an application.

Registering Services#

svcs.pyramid.register_factory(config, svc_type, factory, *, enter=True, ping=None, on_registry_close=None)[source]#

Same as svcs.Registry.register_factory(), but uses registry on config.

svcs.pyramid.register_value(config, svc_type, value, *, enter=False, ping=None, on_registry_close=None)[source]#

Same as svcs.Registry.register_value(), but uses registry on config.

Service Acquisition#

svcs.pyramid.get(request, *svc_types)[source]#

Same as svcs.Container.get(), but uses the container from request.

svcs.pyramid.get_abstract(request, *svc_types)[source]#

Same as svcs.Container.get_abstract(), but uses container from request.

svcs.pyramid.get_pings(request)[source]#

Like svcs.Container.get_pings(), but uses container on request.

See also

Health Checks