Module Index

Module contents

class picodi.AutoScope

Bases: Scope

AutoScope is a scope that automatically closes dependencies after exiting the context.

Don’t use this class directly.

class picodi.ContextVarScope

Bases: ManualScope

ContextVar scope. Values cached in contextvars. Dependencies closed only when user manually call shutdown_dependencies().

class picodi.ManualScope

Bases: Scope

ManualScope is a scope that requires manual closing of dependencies. For example SingletonScope or ContextVarScope use this scope. You can close dependencies by calling shutdown_dependencies() or shutdown_dependencies(scope_class=MyCustomScope) for shutdown only dependencies that uses MyCustomScope scope.

Don’t use this class directly. Inherit this class for your custom scope.

enter(context_manager: AsyncContextManager | ContextManager, *, global_key: Hashable) → Awaitable

Hook for entering yielded dependencies context. Will be called automatically by picodi or when you call init_dependencies().

Parameters:

• context_manager – context manager created from yield dependency.

• global_key – typically a function that requesting dependencies

shutdown(exc: BaseException | None = None, *, global_key: Hashable) → Awaitable

Hook for shutdown dependencies. Will be called when you call shutdown_dependencies()

Parameters:

• exc – exception that was raised in the context.

• global_key – typically a function that requesting dependencies

class picodi.NullScope

Bases: AutoScope

Null scope. Values aren’t cached, dependencies closed automatically after function call. This is the default scope.

picodi.Provide(dependency: Callable[[...], Any], /) → Any

Declare a provider. It takes a single “dependency” callable (like a function). Don’t call it directly, picodi will call it for you.

Parameters:

dependency – can be a regular function or a generator with one yield. If the dependency is a generator, it will be used as a context manager. Any generator that is valid for contextlib.contextmanager() can be used as a dependency.

Example

from picodi import Provide, inject


def get_db():
    yield "db connection"
    print("closing db connection")


@inject
def my_service(db: str = Provide(get_db)):
    assert db == "db connection"

class picodi.Scope

Bases: object

Scopes are used to store and retrieve values by key and for closing dependencies.

Don’t use this class directly and don’t inherit from it. Inherit from AutoScope or ManualScope.

get(key: Hashable, *, global_key: Hashable) → Any

Get a value by key. If value is not exists must raise KeyError.

Parameters:

• key – key to get value, typically a dependency function.

• global_key – typically a function that requesting dependencies

Raises:

KeyError – if value not exists.

set(key: Hashable, value: Any, *, global_key: Hashable) → None

Set a value by key.

Parameters:

• key – key to set value, typically a dependency function.

• value – value to set, typically a dependency instance.

• global_key – typically a function that requesting dependencies

enter_inject(global_key: Hashable) → None

Called when entering an inject() decorator.

Parameters:

global_key – typically a function that requesting dependencies

exit_inject(exc: BaseException | None = None, *, global_key: Hashable) → None

Called before exiting a inject() decorator.

shutdown will be called after this, e.g.:

exit_inject -> shutdown -> inject wrapper returns.

Parameters:

• exc – exception that was raised in the context.

• global_key – typically a function that requesting dependencies

class picodi.SingletonScope

Bases: ManualScope

Singleton scope. Values cached for the lifetime of the application. Dependencies closed only when user manually call shutdown_dependencies().

picodi.inject(fn: Callable[P, T] | None = None, *, registry: Registry | None = None) → Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]

Decorator to inject dependencies into a function. Use it in combination with Provide() to declare dependencies. Should be placed first in the decorator chain (on bottom).

Parameters:

fn – function to decorate.

Example

from picodi import inject, Provide


@inject
def my_service(db=Provide(some_dependency_func)):
    pass

class picodi.Registry(for_init: Iterable[Callable[[...], Any]] | Callable[[], Iterable[Callable[[...], Any]]] | None = None)

Manages dependencies and overrides.

add(dependency: Callable[[...], Any], scope_class: type[AutoScope | ManualScope] = NullScope) → None

Add a dependency to the registry and set scope_class for it.

add_for_init(dependencies: Iterable[Callable[[...], Any]] | Callable[[], Iterable[Callable[[...], Any]]]) → None

Add a dependencies to the list of dependencies to initialize.

set_scope(scope_class: type[AutoScope | ManualScope], *, auto_init: bool = False) → Callable[[TC], TC]

Decorator to declare a dependency. Should be placed last in the decorator chain (on top).

Parameters:

• scope_class – specify the scope class to use it for the dependency.

• auto_init – if set to True, the dependency will be added to the list of dependencies to initialize. This is useful for dependencies that need to be initialized before the application starts.

init(dependencies: Iterable[Callable[[...], Any]] | Callable[[], Iterable[Callable[[...], Any]]] | None = None) → Awaitable

Call this method to init dependencies. Usually, it should be called when your application is starting up.

This method works both for synchronous and asynchronous dependencies. If you call it without await, it will initialize only sync dependencies. If you call it await init(...), it will initialize both sync and async dependencies.

Parameters:

dependencies – dependencies to initialize. If this argument is passed - init dependencies specified in the registry will be ignored.

shutdown(scope_class: type[ManualScope] | tuple[type[ManualScope], ...] = ManualScope) → Awaitable

Call this method to close dependencies. Usually, it should be called when your application is shut down.

This method works both for synchronous and asynchronous dependencies. If you call it without await, it will shutdown only sync dependencies. If you call it await shutdown(), it will shutdown both sync and async dependencies.

If you not pass any arguments, it will shutdown subclasses of ManualScope.

Parameters:

scope_class – you can specify the scope class to shutdown. If passed - only dependencies of this scope class and its subclasses will be shutdown.

lifespan() → Generator[None]

Context manager to manage the lifespan of the application. It will automatically call init and shutdown methods.

alifespan() → AsyncGenerator[None]

Async context manager to manage the lifespan of the application. It will automatically call init and shutdown methods.

property touched: frozenset[Callable[[...], Any]]

Get all dependencies that were used during the picodi lifecycle. This method will return a frozenset of dependencies that were resolved. It will not include dependencies that were overridden. Primarily used for testing purposes. For example, you can check that mongo database was used in the test and clear it after the test.

override(dependency: Callable[[...], Any], new_dependency: Callable[[...], Any] | None) → ContextManager[None]

Override a dependency with a new one. It can be used as a context manager or as a regular method call. New dependency will be added to the registry.

Parameters:

• dependency – dependency to override

• new_dependency – new dependency to use. If explicitly set to None, it will remove the override.

Examples

with registry.override(get_settings, real_settings):
    pass

registry.override(get_settings, real_settings)
registry.override(get_settings, None)  # clear override

resolve(dependency: Callable[[...], Generator[T]]) → ContextManager[T]

resolve(dependency: Callable[[...], T]) → ContextManager[T]

Resolve a dependency synchronously. Returns a context manager that will return the result of the dependency. :param dependency: dependency to resolve. :return: sync context manager.

aresolve(dependency: Callable[[...], Generator[T]]) → AsyncContextManager[T]

aresolve(dependency: Callable[[...], AsyncGenerator[T]]) → AsyncContextManager[T]

aresolve(dependency: Callable[[...], Awaitable[T]]) → AsyncContextManager[T]

aresolve(dependency: Callable[[...], T]) → AsyncContextManager[T]

Resolve a dependency asynchronously. Returns a context manager that will return the result of the dependency. Also can resolve sync dependencies in async context. :param dependency: dependency to resolve. :return: async context manager.

clear_overrides() → None

Clear all overrides. It will remove all overrides, but keep the dependencies.

picodi.registry = <Registry>

Picodi registry. You can use it to register dependencies, scopes, overrides, initialize and shutdown dependencies.

Submodules

picodi.integrations module

Starlette

class picodi.integrations.starlette.RequestScopeMiddleware(app: ASGIApp, *, registry: picodi.Registry | None = None, dependencies_for_init: picodi.InitDependencies | None = None)

Bases: object

Starlette Pure ASGI Middleware for automatically initializing and closing request scoped dependencies

FastAPI

picodi.integrations.fastapi.Provide(dependency: Callable[[...], Any], /, *, wrap: bool = False) → Any

Drop-in replacement for picodi.Provide() but for FastAPI.

Parameters:

• dependency – callable dependency.

• wrap – wrap dependency in fastapi.Depends. In this mode you can use picodi dependencies in FastAPI routes without picodi.inject() decorator.

class picodi.integrations.fastapi.RequestScopeMiddleware(app: ASGIApp, *, registry: picodi.Registry | None = None, dependencies_for_init: picodi.InitDependencies | None = None)

Bases: object

Starlette Pure ASGI Middleware for automatically initializing and closing request scoped dependencies

picodi.helpers module

Helper functions and classes for picodi.

picodi.helpers.get_value(path: str, obj: ~typing.Any, *, default: ~typing.Any = <object object>) → Any

Get attribute from nested objects. Can be useful to avoid passing entire objects like settings to functions that only need a small part of it.

Parameters:

• path – path to the attribute, separated by dots.

• obj – object to search the attribute in.

• default – default value to return if the attribute is not found. If not provided, an PathNotFoundError is raised.

Raises:

PathNotFoundError – if the path is not found in the object and default is not provided.

Example

obj = SimpleNamespace(foo=SimpleNamespace(bar={"baz": 42}))

get_value("foo.bar.baz", obj)  # Output: 42
get_value("foo.bar.baz2", obj)  # Output: AttributeError
get_value("foo.bar.baz2", obj, default=12)  # Output: 12

class picodi.helpers.resolve(dependency: Callable[[], Coroutine[T, None, None] | Generator[T] | AsyncGenerator[T] | T], registry: Registry | None = None)

Bases: Generic[T]

Create a context manager from a dependency.

Can be handy for testing or when you need to create “fabric” dependency (dependency that creates other dependencies based on some conditions).

Parameters:

dependency – dependency to create a context manager from. Like with Provide() - don’t call the dependency function here, just pass it.

Returns:

sync or async context manager.

Example

from picodi.helpers import resolve


def get_42():
    yield 42


with resolve(get_42) as val:
    assert val == 42

picodi.support module

Support module for picodi package.

May be useful for writing your own scopes or other customizations.

class picodi.support.NullAwaitable

Bases: object

Dummy awaitable that does nothing.

class picodi.support.ExitStack

Bases: object

A context manager that combines multiple context managers - both sync and async.

Under the hood, it uses contextlib.ExitStack for sync context managers and contextlib.AsyncExitStack for async

enter_context(cm: AsyncContextManager | ContextManager) → Any

Enters a new context manager and adds its __[a]exit__() method to the callback stack. The return value is the result of the context manager’s own __[a]enter__() method.

Parameters:

cm – context manager to enter.

Returns:

Result of the context manager’s __[a]enter__ method.

close(exc: BaseException | None = None) → Awaitable

Immediately unwinds the callback stack, invoking callbacks in the reverse order of registration.

Parameters:

exc – exception to be passed to the __[a]exit__ method.

picodi.support.is_async_environment() → bool

Check if we are in async environment.

Returns:

True if we are in async environment, False otherwise.

picodi.support.is_async_function(fn: Any) → bool

Check if the function is async.

Parameters:

fn – function to check.

Returns:

True if the function is async, False otherwise.

picodi.support.call_cm_sync(cm: ContextManager) → Any

Call a sync context manager and return its result.

Parameters:

cm – context manager to call.

Returns:

Result of the context manager’s __enter__ method.

async picodi.support.call_cm_async(cm: AsyncContextManager) → Any

Call an async context manager and return its result.

Parameters:

cm – context manager to call.

Returns:

Result of the context manager’s __aenter__ method.