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: 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[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