Skip to content

Guides

Side Effects

To create a provider that does not yield a value but which has side effects, create a subtype of None with the NewType function from the typing module. This new subtype can then be used as a dependency.

from collections.abc import Iterator
from typing import NewType

from pybooster import injector
from pybooster import provider
from pybooster import required
from pybooster import solution

SwitchOn = NewType("SwitchOn", None)


SWITCH = False


@provider.contextmanager
def switch_on() -> Iterator[SwitchOn]:
    global SWITCH
    SWITCH = True
    try:
        yield
    finally:
        SWITCH = False


@injector.function
def is_switch_on(*, _: SwitchOn = required) -> bool:
    return SWITCH


with solution(switch_on):
    assert not SWITCH
    assert is_switch_on()
    assert not SWITCH

Singletons

Eager Singletons

The easiest way to declare a static value is by defining a scope with the new_scope context manager while passing a tuple with the dependency and its value. This will bind the dependency's value to the scope and reuse it whenever the dependency is requested.

from dataclasses import dataclass

from pybooster import injector
from pybooster import new_scope
from pybooster import required


@dataclass
class Dataset:
    x: list[float]
    y: list[float]


@injector.function
def get_dataset(*, dataset: Dataset = required) -> Dataset:
    return dataset


with new_scope({Dataset: Dataset(x=[1, 2, 3], y=[4, 5, 6])}):
    dataset_1 = get_dataset()
    dataset_2 = get_dataset()
    assert dataset_1 is dataset_2

Lazy Singletons

To have a static value that is created at the last possible moment, you'll need to define a solution with a provider that returns the value. Then you'll create a new scope with the new_scope context manager while passing the desired dependency without a value to instead request it from the provider.

from dataclasses import dataclass

from pybooster import injector
from pybooster import new_scope
from pybooster import provider
from pybooster import required
from pybooster import solution


@dataclass
class Dataset:
    x: list[float]
    y: list[float]


count = 0


@provider.function
def dataset_provider() -> Dataset:
    global count
    count += 1
    return Dataset(x=[1, 2, 3], y=[4, 5, 6])


@injector.function
def get_dataset(*, dataset: Dataset = required) -> Dataset:
    return dataset


with solution(dataset_provider):
    dataset_1 = get_dataset()
    dataset_2 = get_dataset()

    # The provider is called twice because the dataset
    # dependency is not active in the current scope.
    assert count == 2
    assert dataset_1 is not dataset_2

    with new_scope(Dataset):
        dataset_3 = get_dataset()
        dataset_4 = get_dataset()

        # The provider is only called one additional time because
        # the dataset dependency is active in the current scope.
        assert count == 3
        assert dataset_3 is dataset_4

Calling Providers

Providers can be called directly as normal context managers with no additional effects.

from typing import NewType

from pybooster import provider

TheAnswer = NewType("TheAnswer", int)


@provider.function
def answer_provider() -> TheAnswer:
    return TheAnswer(42)


with answer_provider() as value:
    assert value == 42

This makes it possible to compose provider implementations without requiring them as a dependency. For example, you could create a SQLAlchemy transaction provider by wrapping a call to the session_provider:

from collections.abc import Iterator
from typing import NewType

from sqlalchemy.orm import Session

from pybooster import provider
from pybooster.extra.sqlalchemy import session_provider

Transaction = NewType("Transaction", Session)


@provider.contextmanager
def transaction_provider() -> Iterator[Transaction]:
    with session_provider() as session, session.begin():
        yield session

Type Hint NameError

Note

This should not be an issue in Python 3.14 with PEP-649.

If you're encountering a NameError when PyBooster tries to infer what type is supplied by a provider or required for an injector this is likely because you're using from __future__ import annotations and the type hint is imported in an if TYPE_CHECKING block. For example, this code raises NameErrors because the Connection type is not present at runtime:

from __future__ import annotations

from contextlib import suppress
from typing import TYPE_CHECKING

from pybooster import injector
from pybooster import provider
from pybooster import required

if TYPE_CHECKING:
    from sqlite3 import Connection


with suppress(NameError):

    @provider.function
    def connection_provider() -> Connection: ...

    raise AssertionError("This should not be reached")


with suppress(NameError):

    @injector.function
    def query_database(*, conn: Connection = required) -> None: ...

    raise AssertionError("This should not be reached")

To fix this, you can move the import outside of the block:

from __future__ import annotations

from sqlite3 import Connection

from pybooster import injector
from pybooster import provider
from pybooster import required


@provider.function
def connection_provider() -> Connection: ...


@injector.function
def query_database(*, conn: Connection = required) -> None: ...

However, some linters like Ruff will automatically move the import back into the block when they discover that the imported value is only used as a type hint. To work around this, you can ignore the linter errors or use the types in such a way that your linter understands they are required at runtime. In the case of Ruff, you'd ignore the following errors:

To convince the linter that types used by PyBooster are required at runtime, you can pass them to the provides argument of the provider decorator or the requires argument of an injector or provider decorator.

from __future__ import annotations

from sqlite3 import Connection

from pybooster import injector
from pybooster import provider
from pybooster import required


@provider.function(provides=Connection)
def connection_provider() -> Connection: ...


@injector.function(requires=[Connection])
def query_database(*, conn: Connection = required) -> None: ...

Tip

Type checkers should still be able to check the return type using the provides argument so it may not be necessary to annotate it in the function signature.

Pytest-Asyncio

Under the hood, PyBooster uses contextvars to manage the state of providers and injectors. If you use pytest-asyncio to write async tests it's likely you'll run into this issue where context established in a fixture is not propagated to your tests. As of writing, the solution is to create a custom event loop policy and task factory as suggested in this comment:

import asyncio
import contextvars
import functools
import traceback
from pathlib import Path

import pytest


def task_factory(loop, coro, context=None):
    stack = traceback.extract_stack()
    for frame in stack[-2::-1]:
        match Path(frame.filename).parts[-2]:
            case "pytest_asyncio":
                break  # use shared context
            case "asyncio":
                pass
            case _:  # create context copy
                context = None
                break
    return asyncio.Task(coro, loop=loop, context=context)


class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    def __init__(self, context) -> None:
        super().__init__()
        self.context = context

    def new_event_loop(self):
        loop = super().new_event_loop()
        loop.set_task_factory(functools.partial(task_factory, context=self.context))
        return loop


@pytest.fixture(scope="session")
def event_loop_policy():
    policy = CustomEventLoopPolicy(contextvars.copy_context())
    yield policy
    policy.get_event_loop().close()