API Specification

prereq.provides(factory: Callable[P, Awaitable[T]], *, level: int | Enum = 1, coverage: list[type[T] | type[Any]] | None = None, cover_parents: bool = True, never_cache: bool = False) AsyncProvider[P, T]
prereq.provides(factory: Callable[P, AsyncIterator[T]], *, level: int | Enum = 1, coverage: list[type[T] | type[Any]] | None = None, cover_parents: bool = True, never_cache: bool = False) AsyncProviderGen[P, T]
prereq.provides(factory: Callable[P, Iterator[T]], *, level: int | Enum = 1, coverage: list[type[T] | type[Any]] | None = None, cover_parents: bool = True, never_cache: bool = False) SyncProviderGen[P, T]
prereq.provides(factory: Callable[P, T], *, level: int | Enum = 1, coverage: list[type[T] | type[Any]] | None = None, cover_parents: bool = True, never_cache: bool = False) SyncProvider[P, T]
prereq.provides(*, level: int | Enum = 1, coverage: list[type[T] | type[Any]] | None = None, cover_parents: bool = True, never_cache: bool = False) _ProviderWrapper[P, T]

Create a provider from factory function or generator.

This is a decorator that also supports being called. For example, you can do:

With defaults.
@provides
def factory(a: A) -> Obj:
    return Obj()

Or you can change the default arguments by calling the decorator.

Changed defaults.
@provides(level=2, never_cache=True)
def factory(a: A) -> Obj:
    return Obj()

This decorator consumes the function. This means that in the previous examples, factory is no longer a function, it would be a SyncProvider. If you’re applying multiple decorators, this one should be the outer-most decorator from the function.

Parameters:
  • factory (Async/Sync function or context-manager compatible generator) – The factory being decorated. If no factory is provided, a decorator is returned.

  • level (int) – The level this provider operates on. For example, a web server may have app level providers (1) and request level providers (2). Default is 1.

  • coverage (list[type[T] | type[Any]] | None) – The types this provider provides. If none are specified, it will be inferred from the factory return notation.

  • cover_parents (bool) – If coverage is inferred, this enables adding coverage for non-abstract inherited types of the return type. Default is True.

  • never_cache (bool) – Disables cacheing the provider in the given resolver scope. Default is False.

Returns:

If the factory is an asynchronous function.

(AsyncProviderGen): If the factory is an asynchronous generator.

(SyncProvider): If the factory is a synchronous function.

(SyncProviderGen): If the factory is a synchronous generator.

Return type:

(AsyncProvider)

class prereq.Resolver(level: int = 1, _parent: ~prereq.resolve.Scope | None = None, _dep_map: dict[int, dict[type[~typing.Any], ~prereq.provide.SyncProvider[~typing.Any, ~typing.Any] | ~prereq.provide.SyncProviderGen[~typing.Any, ~typing.Any] | ~prereq.provide.AsyncProvider[~typing.Any, ~typing.Any] | ~prereq.provide.AsyncProviderGen[~typing.Any, ~typing.Any]]] = <factory>)

The dependency injector.

Resolvers hold on to providers, and create / manage scopes. Resolvers also allow for easy creation of subresolvers. Because scopes are ephemeral and internal, resolvers do not provide any method to access their temporary scopes.

level

The level this resolver operates on. Defaults to level 1.

Type:

int

add_providers(*providers: SyncProvider[Any, Any] | SyncProviderGen[Any, Any] | AsyncProvider[Any, Any] | AsyncProviderGen[Any, Any]) None

Add providers to the resolver.

Providers are functions decorated with prereq.provides(). Providers will be made accessible to resolvers related to this resolver at the same level as the provider.

Parameters:

*providers (All Providers) – Functions decorated with prereq.provides().

resolve(func: Callable[..., Any], cache: dict[type[Any], Any] | None = None) AsyncGenerator[dict[str, Any]]

Create keyword arguements for a function using providers.

This function is an context manager, and should be used as following:

Resolve example.
async with resolver.resolve(func) as kwargs:
    func(**kwargs)

Any untyped arguements, or missing dependencies, will be ignored.

Parameters:
  • func (Callable[..., Any]) – The function to get dependencies for.

  • cache (dict[type[Any], Any], optional) – An optional premade cache to set constant values. Defaults to None.

Yields:

(dict[str, Any]) – The dependencies for the function’s parameters, except for non-typed parameters.

All Providers

class prereq.AsyncProvider(coverage: Iterable[type[T] | type[Any]], args: dict[str, type[Any]], level: int, never_cache: bool, factory: F)

Wrap an asynchronous provider function.

coverage

The types this provider handles.

Type:

Iterable[type[T | Any]]

args

The provider’s arguement types.

Type:

dict[str, type[Any]]

level

The level this provider operates on.

Type:

int

never_cache

If the provider should be cached after being called.

Type:

bool

factory

The function wrapped by this provider.

Type:

F

class prereq.AsyncProviderGen(coverage: Iterable[type[T] | type[Any]], args: dict[str, type[Any]], level: int, never_cache: bool, factory: F)

Wrap an asynchronous context provider function.

coverage

The types this provider handles.

Type:

Iterable[type[T | Any]]

args

The provider’s arguement types.

Type:

dict[str, type[Any]]

level

The level this provider operates on.

Type:

int

never_cache

If the provider should be cached after being called.

Type:

bool

factory

The async generator which will become a context manager.

Type:

F

class prereq.SyncProvider(coverage: Iterable[type[T] | type[Any]], args: dict[str, type[Any]], level: int, never_cache: bool, factory: F)

Wrap a synchronous provider function.

coverage

The types this provider handles.

Type:

Iterable[type[T | Any]]

args

The provider’s arguement types.

Type:

dict[str, type[Any]]

level

The level this provider operates on.

Type:

int

never_cache

If the provider should be cached after being called.

Type:

bool

factory

The function wrapped by this provider.

Type:

F

class prereq.SyncProviderGen(coverage: Iterable[type[T] | type[Any]], args: dict[str, type[Any]], level: int, never_cache: bool, factory: F)

Wrap a synchronous context provider function.

coverage

The types this provider handles.

Type:

Iterable[type[T | Any]]

args

The provider’s arguement types.

Type:

dict[str, type[Any]]

level

The level this provider operates on.

Type:

int

never_cache

If the provider should be cached after being called.

Type:

bool

factory

The generator which will become a context manager.

Type:

F

Internal

class prereq.resolve.Scope(parent: Scope | None, level: int, providers: MappingProxyType)

Temporary cache and context manager.

Scopes are created by Resolvers to manage a specific level’s cache, and close context managers. A rounter creates a scope when initiating a sub router, as well as when it’s resolving a function’s params. Scopes do not themselves ensure that prereq.resolve.Scope.cleanup() gets called. That should bew handled by the process using the scope.

parent

The previous level’s scope, if applicable.

Type:

Scope | None

level

The level this scope services.

Type:

int

providers

Providers on this scope’s level.

Type:

MappingProxyType[type[Any], Providers]

cache

Instances of types which have already been gathered for this scope.

Type:

dict[type[Any], Any]

async cleanup() None

Resolve pending context managers.

Some providers are context managers that need to be exited when the scope is left. This method exits all pending context managers created by the scope.

async get(typ: type[T]) T

Create an instance of a type using a provider.

This method will check this scope’s providers to see if any support this type. If none do, it will pass the request off to the parent scope. The level associated with that type will then cache the result, if the provide has caching enabled.

Parameters:

typ (type[T]) – The type that needs to be instantiated by a provider.

Returns:

Returns an instance of the requested type.

Return type:

T

Raises:

ProviderNotFoundError – If no provider is found, and this scope has no parent, this error will be raised.