Usage¶

svcs-pyramid 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.

Installation¶

$ uv pip install svcs-pyramid

Note

svcs-pyramid used to be the svcs.pyramid module that shipped with svcs itself. If you’re migrating, the only change you have to make is the import:

import svcs           →  import svcs_pyramid
svcs.pyramid.init(…)  →  svcs_pyramid.init(…)

The names under which the registry and the container are stored are unchanged, so both packages are interchangeable.

Initialization¶

The most important API 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_pyramid


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¶

You can use svcs_pyramid.get_pings() to get all registered health checks.

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_pyramid


@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_pyramid
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.