Reference

pybooster

pybooster.required

required = make_sentinel_value(__name__, 'required')

A sentinel object used to indicate that a dependency is required.

pybooster.solved

solved(*providers)

Resolve the dependency graph defined by the given providers during the context.

Parameters:

Name	Type	Description	Default
providers	Provider[[], Any] | Sequence[Provider[[], Any]]	The providers that define the dependency graph to be resolved given as positional arguments or as sequences of providers.	()

Source code in src/pybooster/core/solution.py

@contextmanager
def solved(*providers: Provider[[], Any] | Sequence[Provider[[], Any]]) -> Iterator[None]:
    """Resolve the dependency graph defined by the given providers during the context.

    Args:
        providers:
            The providers that define the dependency graph to be resolved given
            as positional arguments or as sequences of providers.
    """
    if not providers:
        msg = "At least one provider must be given."
        raise ValueError(msg)
    sync_infos: dict[type, SyncProviderInfo] = {}
    async_infos: dict[type, AsyncProviderInfo] = {}
    for p in _normalize_providers(providers):
        if isinstance(p, SyncProvider):
            sync_infos.update(
                get_provider_info(p.producer, p.provides, p.dependencies, is_sync=True)
            )
        else:
            async_infos.update(
                get_provider_info(p.producer, p.provides, p.dependencies, is_sync=False)
            )
    reset = set_solutions(sync_infos, async_infos, _CURRENT_VALUES.get().keys())
    try:
        yield
    finally:
        reset()

pybooster.core

pybooster.core.injector

pybooster.core.injector.required

required = make_sentinel_value(__name__, 'required')

A sentinel object used to indicate that a dependency is required.

pybooster.core.injector.CurrentValues

Bases: Mapping[Hint, Any]

A mapping from dependency types to their current values.

pybooster.core.injector.asynccontextmanager

asynccontextmanager(func, *, requires=None, shared=False)

Inject dependencies into the given async context manager function.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of the context.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asynccontextmanager(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, AbstractAsyncContextManager[R]]:
    """Inject dependencies into the given async context manager function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of the context.
    """
    return _asynccontextmanager(asynciterator(func, requires=requires, shared=shared))

pybooster.core.injector.asyncfunction

asyncfunction(func, *, requires=None, shared=False)

Inject dependencies into the given coroutine.

Parameters:

Name	Type	Description	Default
func	Callable[P, Coroutine[Any, Any, R]]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asyncfunction(
    func: Callable[P, Coroutine[Any, Any, R]],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, Coroutine[Any, Any, R]]:
    """Inject dependencies into the given coroutine.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:  # type: ignore[reportReturnType]
        stack = AsyncFastStack()
        try:
            await async_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            return await func(*args, **kwargs)
        finally:
            await stack.aclose()

    return wrapper

pybooster.core.injector.asynciterator

asynciterator(func, *, requires=None, shared=False)

Inject dependencies into the given async iterator.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asynciterator(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> AsyncIteratorCallable[P, R]:
    """Inject dependencies into the given async iterator.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[R]:
        stack = AsyncFastStack()
        try:
            await async_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            async for value in func(*args, **kwargs):
                yield value
        finally:
            await stack.aclose()

    return wrapper

pybooster.core.injector.contextmanager

contextmanager(func, *, requires=None, shared=False)

Inject dependencies into the given context manager function.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of the context.	False

Source code in src/pybooster/core/injector.py

@paramorator
def contextmanager(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, AbstractContextManager[R]]:
    """Inject dependencies into the given context manager function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of the context.
    """
    return _contextmanager(iterator(func, requires=requires, shared=shared))

pybooster.core.injector.current_values

current_values()

Get a mapping from dependency types to their current values.

Source code in src/pybooster/core/injector.py

def current_values() -> CurrentValues:
    """Get a mapping from dependency types to their current values."""
    return cast("CurrentValues", dict(_CURRENT_VALUES.get()))

pybooster.core.injector.function

function(func, *, requires=None, shared=False)

Inject dependencies into the given function.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def function(
    func: Callable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, R]:
    """Inject dependencies into the given function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        stack = FastStack()
        try:
            sync_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            return func(*args, **kwargs)
        finally:
            stack.close()

    return wrapper

pybooster.core.injector.iterator

iterator(func, *, requires=None, shared=False)

Inject dependencies into the given iterator.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def iterator(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> IteratorCallable[P, R]:
    """Inject dependencies into the given iterator.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterator[R]:
        stack = FastStack()
        try:
            sync_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            yield from func(*args, **kwargs)
        finally:
            stack.close()

    return wrapper

pybooster.core.injector.shared

shared(*args)

Share the values for a set of dependencies for the duration of a context.

Source code in src/pybooster/core/injector.py

def shared(*args: Hint | tuple[Hint, Any]) -> _SharedContext:
    """Share the values for a set of dependencies for the duration of a context."""
    param_vals: dict[str, Any] = {}
    param_deps: dict[str, Hint] = {}
    for index, arg in enumerate(args):
        key = f"__{index}"
        match arg:
            case [cls, val]:
                param_vals[key] = val
                param_deps[key] = cls
            case cls:
                param_deps[key] = cls
    return _SharedContext(param_vals, param_deps)

pybooster.core.provider

pybooster.core.provider.Provider

Provider = 'SyncProvider[P, R] | AsyncProvider[P, R]'

A provider for a dependency.

pybooster.core.provider.AsyncProvider

AsyncProvider(producer, provides, dependencies)

Bases: Generic[P, R], _BaseProvider[R]

A provider for a dependency.

Source code in src/pybooster/core/provider.py

def __init__(
    self,
    producer: AsyncContextManagerCallable[P, R],
    provides: Hint | InferHint,
    dependencies: HintMap,
) -> None:
    self.producer = producer
    self.provides = provides
    self.dependencies = dependencies

pybooster.core.provider.AsyncProvider.bind

bind(*args, **kwargs)

Inject the dependencies and produce the dependency.

Source code in src/pybooster/core/provider.py

def bind(self, *args: P.args, **kwargs: P.kwargs) -> AsyncProvider[[], R]:
    """Inject the dependencies and produce the dependency."""
    ...

pybooster.core.provider.SyncProvider

SyncProvider(producer, provides, dependencies)

Bases: Generic[P, R], _BaseProvider[R]

A provider for a dependency.

Source code in src/pybooster/core/provider.py

def __init__(
    self,
    producer: ContextManagerCallable[P, R],
    provides: Hint | InferHint,
    dependencies: HintMap,
) -> None:
    self.producer = producer
    self.provides = provides
    self.dependencies = dependencies

pybooster.core.provider.SyncProvider.bind

bind(*args, **kwargs)

Inject the dependencies and produce the dependency.

Source code in src/pybooster/core/provider.py

def bind(self, *args: P.args, **kwargs: P.kwargs) -> SyncProvider[[], R]:
    """Inject the dependencies and produce the dependency."""
    ...

pybooster.core.provider.asyncfunction

asyncfunction(func, *, requires=None, provides=None)

Create a provider from the given coroutine.

Parameters:

Name	Type	Description	Default
func	Callable[P, Awaitable[R]]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def asyncfunction(
    func: Callable[P, Awaitable[R]],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> AsyncProvider[P, R]:
    """Create a provider from the given coroutine.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_coroutine_return_type(func)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[R]:
        yield await func(*args, **kwargs)

    return asynciterator(wrapper, provides=provides, requires=requires)

pybooster.core.provider.asynciterator

asynciterator(func, *, requires=None, provides=None)

Create a provider from the given async iterator function.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def asynciterator(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> AsyncProvider[P, R]:
    """Create a provider from the given async iterator function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_iterator_yield_type(func, sync=False)
    requires = get_required_parameters(func, requires)
    return AsyncProvider(_asynccontextmanager(func), cast("type[R]", provides), requires)

pybooster.core.provider.function

function(func, *, requires=None, provides=None)

Create a provider from the given function.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def function(
    func: Callable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> SyncProvider[P, R]:
    """Create a provider from the given function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_callable_return_type(func)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterator[R]:
        yield func(*args, **kwargs)

    return iterator(wrapper, provides=provides, requires=requires)

pybooster.core.provider.iterator

iterator(func, *, requires=None, provides=None)

Create a provider from the given iterator function.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def iterator(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> SyncProvider[P, R]:
    """Create a provider from the given iterator function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_iterator_yield_type(func, sync=True)
    requires = get_required_parameters(func, requires)
    return SyncProvider(_contextmanager(func), cast("type[R]", provides), requires)

pybooster.core.solution

pybooster.core.solution.solved

solved(*providers)

Resolve the dependency graph defined by the given providers during the context.

Parameters:

Name	Type	Description	Default
providers	Provider[[], Any] | Sequence[Provider[[], Any]]	The providers that define the dependency graph to be resolved given as positional arguments or as sequences of providers.	()

Source code in src/pybooster/core/solution.py

@contextmanager
def solved(*providers: Provider[[], Any] | Sequence[Provider[[], Any]]) -> Iterator[None]:
    """Resolve the dependency graph defined by the given providers during the context.

    Args:
        providers:
            The providers that define the dependency graph to be resolved given
            as positional arguments or as sequences of providers.
    """
    if not providers:
        msg = "At least one provider must be given."
        raise ValueError(msg)
    sync_infos: dict[type, SyncProviderInfo] = {}
    async_infos: dict[type, AsyncProviderInfo] = {}
    for p in _normalize_providers(providers):
        if isinstance(p, SyncProvider):
            sync_infos.update(
                get_provider_info(p.producer, p.provides, p.dependencies, is_sync=True)
            )
        else:
            async_infos.update(
                get_provider_info(p.producer, p.provides, p.dependencies, is_sync=False)
            )
    reset = set_solutions(sync_infos, async_infos, _CURRENT_VALUES.get().keys())
    try:
        yield
    finally:
        reset()

pybooster.core.state

pybooster.core.state.copy_state

copy_state()

Copy PyBooster's current state and return a callback that will set it in another context.

Example

If you need to run a function in a different context, you can use this function to copy the state established there and set it in the new context. This might happen if you create a thread (which has its own context) to run a function that creates some PyBooster state which you'd then like to use in the main thread.

from concurrent.futures import Future
from threading import Thread
from typing import NewType

from pybooster import injector
from pybooster.core.state import copy_state

Greeting = NewType("Greeting", str)


def from_thread(future):
    with injector.shared((Greeting, "Hello")):
        set_state = copy_state()
        future.set_result(set_state)


set_state_future = Future()
thread = Thread(target=from_thread, args=(set_state_future,))
thread.start()
set_state = set_state_future.result()

reset_state = set_state()
assert injector.current_values().get(Greeting) == "Hello"

reset_state()
assert injector.current_values().get(Greeting) is None

Source code in src/pybooster/core/state.py

def copy_state() -> StateSetter:
    """Copy PyBooster's current state and return a callback that will set it in another context.

    Example:
        If you need to run a function in a different context, you can use this function
        to copy the state established there and set it in the new context. This might
        happen if you create a thread (which has its own context) to run a function that
        creates some PyBooster state which you'd then like to use in the main thread.

        ```python
        from concurrent.futures import Future
        from threading import Thread
        from typing import NewType

        from pybooster import injector
        from pybooster.core.state import copy_state

        Greeting = NewType("Greeting", str)


        def from_thread(future):
            with injector.shared((Greeting, "Hello")):
                set_state = copy_state()
                future.set_result(set_state)


        set_state_future = Future()
        thread = Thread(target=from_thread, args=(set_state_future,))
        thread.start()
        set_state = set_state_future.result()

        reset_state = set_state()
        assert injector.current_values().get(Greeting) == "Hello"

        reset_state()
        assert injector.current_values().get(Greeting) is None
        ```
    """
    current_values = _CURRENT_VALUES.get()
    full_solution = _FULL_SOLUTION.get()
    sync_solution = _SYNC_SOLUTION.get()

    def set_state() -> StateResetter:
        current_values_token = _CURRENT_VALUES.set(current_values)
        full_solution_token = _FULL_SOLUTION.set(full_solution)
        sync_solution_token = _SYNC_SOLUTION.set(sync_solution)

        def reset_state() -> None:
            _SYNC_SOLUTION.reset(sync_solution_token)
            _FULL_SOLUTION.reset(full_solution_token)
            _CURRENT_VALUES.reset(current_values_token)

        return reset_state

    return set_state

pybooster.extra

pybooster.extra.asgi

pybooster.extra.asgi.PyBoosterMiddleware

PyBoosterMiddleware(app)

ASGI middleware to manage PyBooster's internal state.

Source code in src/pybooster/extra/asgi.py

def __init__(self, app: Asgi) -> None:
    self.app = app

pybooster.extra.sqlalchemy

pybooster.extra.sqlalchemy.async_session_provider

async_session_provider = asynciterator(
    _async_session_provider,
    requires={"bind": AsyncEngine},
    provides=_infer_async_session_type,
)

Provide an async SQLAlchemy session.

pybooster.extra.sqlalchemy.session_provider

session_provider = iterator(
    _session_provider,
    requires={"bind": Engine},
    provides=_infer_session_type,
)

Provide a SQLAlchemy session.

pybooster.extra.sqlalchemy.AsyncSessionMaker

Bases: Protocol[P, A_co]

A protocol for creating an async SQLAlchemy session.

pybooster.extra.sqlalchemy.AsyncSessionMaker.__call__

__call__(bind=..., *args, **kwargs)

Create an async SQLAlchemy session.

Source code in src/pybooster/extra/sqlalchemy.py

def __call__(self, bind: AsyncEngine = ..., *args: P.args, **kwargs: P.kwargs) -> A_co:
    """Create an async SQLAlchemy session."""
    ...

pybooster.extra.sqlalchemy.SessionMaker

Bases: Protocol[P, S_co]

A protocol for creating a SQLAlchemy session.

pybooster.extra.sqlalchemy.SessionMaker.__call__

__call__(bind=..., *args, **kwargs)

Create a SQLAlchemy session.

Source code in src/pybooster/extra/sqlalchemy.py

def __call__(self, bind: Engine = ..., *args: P.args, **kwargs: P.kwargs) -> S_co:
    """Create a SQLAlchemy session."""
    ...

pybooster.extra.sqlalchemy.async_engine_provider

async_engine_provider(*args, **kwargs)

Provide an async SQLAlchemy engine.

Source code in src/pybooster/extra/sqlalchemy.py

@provider.asynciterator
@copy_signature(create_async_engine)
async def async_engine_provider(*args: Any, **kwargs: Any) -> AsyncIterator[AsyncEngine]:
    """Provide an async SQLAlchemy engine."""
    engine = create_async_engine(*args, **kwargs)
    try:
        yield engine
    finally:
        await engine.dispose()

pybooster.extra.sqlalchemy.engine_provider

engine_provider(*args, **kwargs)

Provide a SQLAlchemy engine.

Source code in src/pybooster/extra/sqlalchemy.py

@provider.iterator
@copy_signature(create_engine)
def engine_provider(*args: Any, **kwargs: Any) -> Iterator[Engine]:
    """Provide a SQLAlchemy engine."""
    engine = create_engine(*args, **kwargs)
    try:
        yield engine
    finally:
        engine.dispose()

pybooster.injector

pybooster.injector.required

required = make_sentinel_value(__name__, 'required')

A sentinel object used to indicate that a dependency is required.

pybooster.injector.CurrentValues

Bases: Mapping[Hint, Any]

A mapping from dependency types to their current values.

pybooster.injector.asynccontextmanager

asynccontextmanager(func, *, requires=None, shared=False)

Inject dependencies into the given async context manager function.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of the context.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asynccontextmanager(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, AbstractAsyncContextManager[R]]:
    """Inject dependencies into the given async context manager function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of the context.
    """
    return _asynccontextmanager(asynciterator(func, requires=requires, shared=shared))

pybooster.injector.asyncfunction

asyncfunction(func, *, requires=None, shared=False)

Inject dependencies into the given coroutine.

Parameters:

Name	Type	Description	Default
func	Callable[P, Coroutine[Any, Any, R]]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asyncfunction(
    func: Callable[P, Coroutine[Any, Any, R]],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, Coroutine[Any, Any, R]]:
    """Inject dependencies into the given coroutine.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:  # type: ignore[reportReturnType]
        stack = AsyncFastStack()
        try:
            await async_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            return await func(*args, **kwargs)
        finally:
            await stack.aclose()

    return wrapper

pybooster.injector.asynciterator

asynciterator(func, *, requires=None, shared=False)

Inject dependencies into the given async iterator.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def asynciterator(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> AsyncIteratorCallable[P, R]:
    """Inject dependencies into the given async iterator.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[R]:
        stack = AsyncFastStack()
        try:
            await async_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            async for value in func(*args, **kwargs):
                yield value
        finally:
            await stack.aclose()

    return wrapper

pybooster.injector.contextmanager

contextmanager(func, *, requires=None, shared=False)

Inject dependencies into the given context manager function.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of the context.	False

Source code in src/pybooster/core/injector.py

@paramorator
def contextmanager(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, AbstractContextManager[R]]:
    """Inject dependencies into the given context manager function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of the context.
    """
    return _contextmanager(iterator(func, requires=requires, shared=shared))

pybooster.injector.current_values

current_values()

Get a mapping from dependency types to their current values.

Source code in src/pybooster/core/injector.py

def current_values() -> CurrentValues:
    """Get a mapping from dependency types to their current values."""
    return cast("CurrentValues", dict(_CURRENT_VALUES.get()))

pybooster.injector.function

function(func, *, requires=None, shared=False)

Inject dependencies into the given function.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def function(
    func: Callable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> Callable[P, R]:
    """Inject dependencies into the given function.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        stack = FastStack()
        try:
            sync_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            return func(*args, **kwargs)
        finally:
            stack.close()

    return wrapper

pybooster.injector.iterator

iterator(func, *, requires=None, shared=False)

Inject dependencies into the given iterator.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to inject dependencies into.	required
requires	HintMap | HintSeq | None	The parameters and dependencies to inject. Otherwise infered from signature.	None
shared	bool	Whether injected values should be shared for the duration of any calls.	False

Source code in src/pybooster/core/injector.py

@paramorator
def iterator(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    shared: bool = False,
) -> IteratorCallable[P, R]:
    """Inject dependencies into the given iterator.

    Args:
        func: The function to inject dependencies into.
        requires: The parameters and dependencies to inject. Otherwise infered from signature.
        shared: Whether injected values should be shared for the duration of any calls.
    """
    requires = get_required_parameters(func, requires)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterator[R]:
        stack = FastStack()
        try:
            sync_inject_into_params(stack, kwargs, requires, keep_current_values=shared)
            yield from func(*args, **kwargs)
        finally:
            stack.close()

    return wrapper

pybooster.injector.shared

shared(*args)

Share the values for a set of dependencies for the duration of a context.

Source code in src/pybooster/core/injector.py

def shared(*args: Hint | tuple[Hint, Any]) -> _SharedContext:
    """Share the values for a set of dependencies for the duration of a context."""
    param_vals: dict[str, Any] = {}
    param_deps: dict[str, Hint] = {}
    for index, arg in enumerate(args):
        key = f"__{index}"
        match arg:
            case [cls, val]:
                param_vals[key] = val
                param_deps[key] = cls
            case cls:
                param_deps[key] = cls
    return _SharedContext(param_vals, param_deps)

pybooster.provider

pybooster.provider.Provider

Provider = 'SyncProvider[P, R] | AsyncProvider[P, R]'

A provider for a dependency.

pybooster.provider.AsyncProvider

AsyncProvider(producer, provides, dependencies)

Bases: Generic[P, R], _BaseProvider[R]

A provider for a dependency.

Source code in src/pybooster/core/provider.py

def __init__(
    self,
    producer: AsyncContextManagerCallable[P, R],
    provides: Hint | InferHint,
    dependencies: HintMap,
) -> None:
    self.producer = producer
    self.provides = provides
    self.dependencies = dependencies

pybooster.provider.AsyncProvider.bind

bind(*args, **kwargs)

Inject the dependencies and produce the dependency.

Source code in src/pybooster/core/provider.py

def bind(self, *args: P.args, **kwargs: P.kwargs) -> AsyncProvider[[], R]:
    """Inject the dependencies and produce the dependency."""
    ...

pybooster.provider.SyncProvider

SyncProvider(producer, provides, dependencies)

Bases: Generic[P, R], _BaseProvider[R]

A provider for a dependency.

Source code in src/pybooster/core/provider.py

def __init__(
    self,
    producer: ContextManagerCallable[P, R],
    provides: Hint | InferHint,
    dependencies: HintMap,
) -> None:
    self.producer = producer
    self.provides = provides
    self.dependencies = dependencies

pybooster.provider.SyncProvider.bind

bind(*args, **kwargs)

Inject the dependencies and produce the dependency.

Source code in src/pybooster/core/provider.py

def bind(self, *args: P.args, **kwargs: P.kwargs) -> SyncProvider[[], R]:
    """Inject the dependencies and produce the dependency."""
    ...

pybooster.provider.asyncfunction

asyncfunction(func, *, requires=None, provides=None)

Create a provider from the given coroutine.

Parameters:

Name	Type	Description	Default
func	Callable[P, Awaitable[R]]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def asyncfunction(
    func: Callable[P, Awaitable[R]],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> AsyncProvider[P, R]:
    """Create a provider from the given coroutine.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_coroutine_return_type(func)

    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[R]:
        yield await func(*args, **kwargs)

    return asynciterator(wrapper, provides=provides, requires=requires)

pybooster.provider.asynciterator

asynciterator(func, *, requires=None, provides=None)

Create a provider from the given async iterator function.

Parameters:

Name	Type	Description	Default
func	AsyncIteratorCallable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def asynciterator(
    func: AsyncIteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> AsyncProvider[P, R]:
    """Create a provider from the given async iterator function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_iterator_yield_type(func, sync=False)
    requires = get_required_parameters(func, requires)
    return AsyncProvider(_asynccontextmanager(func), cast("type[R]", provides), requires)

pybooster.provider.function

function(func, *, requires=None, provides=None)

Create a provider from the given function.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def function(
    func: Callable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> SyncProvider[P, R]:
    """Create a provider from the given function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_callable_return_type(func)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterator[R]:
        yield func(*args, **kwargs)

    return iterator(wrapper, provides=provides, requires=requires)

pybooster.provider.iterator

iterator(func, *, requires=None, provides=None)

Create a provider from the given iterator function.

Parameters:

Name	Type	Description	Default
func	IteratorCallable[P, R]	The function to create a provider from.	required
requires	HintMap | HintSeq | None	The dependencies of the function (infered if not provided).	None
provides	Hint | InferHint | None	The type that the function provides (infered if not provided).	None

Source code in src/pybooster/core/provider.py

@paramorator
def iterator(
    func: IteratorCallable[P, R],
    *,
    requires: HintMap | HintSeq | None = None,
    provides: Hint | InferHint | None = None,
) -> SyncProvider[P, R]:
    """Create a provider from the given iterator function.

    Args:
        func: The function to create a provider from.
        requires: The dependencies of the function (infered if not provided).
        provides: The type that the function provides (infered if not provided).
    """
    provides = provides or get_iterator_yield_type(func, sync=True)
    requires = get_required_parameters(func, requires)
    return SyncProvider(_contextmanager(func), cast("type[R]", provides), requires)

pybooster.types

pybooster.types.AnyContextManagerCallable

AnyContextManagerCallable = Callable[
    P,
    AbstractContextManager[R]
    | AbstractAsyncContextManager[R],
]

A callable that returns any kind of context manager.

pybooster.types.AsyncContextManagerCallable

AsyncContextManagerCallable = Callable[
    P, AbstractAsyncContextManager[R]
]

A callable that returns an async context manager.

pybooster.types.AsyncIteratorCallable

AsyncIteratorCallable = Callable[P, AsyncIterator[R]]

A callable that returns an async iterator.

pybooster.types.ContextManagerCallable

ContextManagerCallable = Callable[
    P, AbstractContextManager[R]
]

A callable that returns a context manager.

pybooster.types.Hint

Hint = type | Any

A type hint.

pybooster.types.HintDict

HintDict = dict[str, Hint]

A dictionary of parameter or attribute names to their type.

pybooster.types.HintMap

HintMap = Mapping[str, Hint]

A mapping of parameter or attribute names to their type.

pybooster.types.HintSeq

HintSeq = Sequence[Hint]

A sequence of types.

pybooster.types.InferHint

InferHint = Callable[..., Hint]

A callable that infers a type hint.

pybooster.types.IteratorCallable

IteratorCallable = Callable[P, Iterator[R]]

A callable that returns an iterator.

pybooster.types.InjectionError

Bases: RuntimeError

An error raised when an injection fails.

pybooster.types.SolutionError

Bases: RuntimeError

An error raised when a solution fails.