Advanced usage

Lifespan

Library allows to create “lifespan” dependencies that can clean-up some resources after data they returned was used

from collections.abc import Generator

from fundi import from_, inject, scan, Scope


def require_session() -> Generator[str, None, None]:
    print("Session set-up")
    yield "session"
    print("Session clean-up")


def application(session: str = from_(require_session)):
    print(f"Application started with {session = }")


inject(Scope(), scan(application))

Also, as lifespan dependencies can be used classes that implement context manager protocol - sync (__enter__ and __exit__) or async (__aenter__ and __aexit__)

from fundi import from_, inject, scan, Scope


class RequireSession:
    def __enter__(self):
        print("Session set-up")
        return "session"

    def __exit__(self, *_):
        print("Session tear-down")
        return False


def application(session: str = from_(RequireSession)):
    print(f"Application started with {session = }")


inject(Scope(), scan(application))

Alternatively, you can use “virtual” context managers.

“Virtual” context managers are drop-in replacements for decorators from the contextlib module, such as @contextmanager or @asynccontextmanager.

from fundi import from_, inject, scan, virtual_context, Scope


@virtual_context
def virtual_ctx():
    print("Virtual context manager set-up")
    yield "Virtual context manager value"
    print("Virtual context manager tear-down")


def application(virtual_ctx_value: str = from_(virtual_ctx)):
    print(f"Application started with {virtual_ctx_value = }")


inject(Scope(), scan(application))

The @virtual_context decorator automatically chooses the correct context manager - asynchronous or synchronous depending on the decorated function type.

Lifespan exception awareness

Lifespan dependencies aware about downstream exceptions. This means you can catch exception that happened during injection in lifespan dependency to do additional cleanup if exception occurred.

Note: Even that lifespan dependency can catch exception does not mean it can ignore it. FunDI does not allow lifespan dependencies to ignore exceptions. So, exception will be re-raised even if lifespan dependency ignored it.

from fundi import from_, scan, inject, injection_trace, Scope


def lifespan():
    try:
        yield
        print("Injection happened successfully")
    except ConnectionRefusedError as e:  # <== Lifespan dependency caught exception
        print("exception happened on teardown:", e)
        print("injection trace:")
        trace = injection_trace(e)
        while trace:
            print(" ", trace.info.call, "with", trace.values)
            trace = trace.origin

        print()


def require_random_animal(b=from_(lifespan)) -> str:
    raise ConnectionRefusedError(
        "Cannot connect to random.animal.com"
    )  # <== Exception happened here


def application(
    animal: str = from_(require_random_animal),
):
    print("Animal:", animal)


try:
    inject(Scope(), scan(application))
except (
    ConnectionRefusedError
):  # <== Lifespan dependency does not reraise exception, but it still goes downstream
    print("ConnectionRefusedError happened on injection")

Own exit stack

If you need to do some work before injected dependencies will tear-down - you can pass your own exit stack to inject() and ainject() functions.

Synchronous example:

import time
from contextlib import ExitStack

from fundi import inject, scan, from_, Scope


def dependency(value: str = "default"):
    print("set-up")
    yield value
    print("tear-down")


def dependant(value: str = from_(dependency)):
    yield "Dependant got " + value


with ExitStack() as stack:
    value = inject(Scope({"value": "value"}), scan(dependant), stack)
    print(value)
    print("Doing some computations before dependencies would tear-down")
    time.sleep(0.8)  # simulate computations

Asynchronous example:

import asyncio
from contextlib import AsyncExitStack

from fundi import inject, scan, from_, Scope


def dependency(value: str = "default"):
    print("set-up")
    yield value
    print("tear-down")


def dependant(value: str = from_(dependency)):
    yield "Dependant got " + value


async def main():
    async with AsyncExitStack() as stack:
        value = inject(Scope({"value": "value"}), scan(dependant), stack)
        print(value)
        print("Doing some computations before dependencies would tear-down")
        await asyncio.sleep(0.8)  # simulate computations


if __name__ == "__main__":
    asyncio.run(main())

Caching

FunDI caches dependency results by default — so each dependency is only evaluated once per injection cycle, avoiding duplicate work or inconsistent data.

from collections.abc import Generator

from fundi import from_, inject, scan, Scope


class Session:
    pass


def require_session() -> Generator[Session, None, None]:
    print("Session set-up")
    yield Session()
    print("Session clean-up")


def intermediate_dependency(session: Session = from_(require_session)):
    return session


def application(
    session: Session = from_(require_session), session1: Session = from_(intermediate_dependency)
):
    assert session is session1
    print(f"Application started with {session = }")


inject(Scope(), scan(application))

To disable this behavior - use caching=False parameter when defining dependant’s dependency:

from collections.abc import Generator
from fundi import from_, inject, scan, Scope


class Session:
    pass


def require_session() -> Generator[Session, None, None]:
    print("Session set-up")
    yield Session()
    print("Session clean-up")


def intermediate_dependency(session: Session = from_(require_session)):
    return session


def application(
    session: Session = from_(intermediate_dependency),
    session1: Session = from_(require_session, caching=False),
):
    # session will be stored in cache and fetched on next occurrences
    # session1 will not be cached nor fetched from cache

    assert session is not session1
    print(f"Application started with {session = }")


inject(Scope(), scan(application))

Scope

Library provides injection scope, that allows to inject values to dependencies parameters by name

from collections.abc import Generator

from fundi import from_, inject, scan, Scope


class Session:
    pass


def require_session(database_url: str) -> Generator[Session, None, None]:
    print(f"Session set-up with {database_url = }")
    yield Session()
    print("Session clean-up")


def application(
    session: Session = from_(require_session),
):
    print(f"Application started with {session = }")


# "database_url" key goes to "database_url" parameter of require_session function
inject(Scope({"database_url": "url"}), scan(application))

Dependency parameter awareness

Dependant’s dependencies know of the parameter they are injected to.

Note: Parameter-aware dependencies are cached just like any other dependency by default.

This means that even if the same function is injected into multiple parameters (e.g., user_id, client_id), it will only be called once, and the cached result will be reused — regardless of which parameter it’s injected into.

If your function depends on the parameter name or annotation (e.g. to extract different headers), you must disable caching manually using from_(..., caching=False).

This behavior is intentional for now and may change in future versions, but currently it’s the developer’s responsibility to manage it.

This can be used to create more transparent dependencies:

from fundi import scan, FromType, from_, inject, Parameter, Scope


def header(param: FromType[Parameter]) -> str:
    print("Parameter name:", param.name)
    print("Parameter annotation:", param.annotation)

    assert param.name == "token"
    assert param.annotation is str

    return f"{param.name}-{param.annotation!r}"


def application(token: str = from_(header)):
    print("Token:", token)

    assert token == f"token-{str!r}"


inject(Scope(), scan(application))

Scope by type

Dependency parameters can resolve their values from scope by type using from_

from fundi import inject, scan, FromType, Scope, Type


class Session:
    pass


def application(
    session: FromType[Session],
):
    print(f"Application started with {session = }")


# Scope key goes to "session" parameter of application function
inject(Scope({Session: Type.instance(Session())}), scan(application))

Exception tracing

FunDI adds injection trace to all exceptions on injection to help you understand them

from fundi import from_, scan, inject, injection_trace, Scope


def require_random_animal() -> str:
    raise ConnectionRefusedError("Failed to connect to server :<")
    return random.choice(["cat", "dog", "chicken", "horse", "platypus", "cow"])


def application(
    animal: str = from_(require_random_animal),
):
    print("Animal:", animal)


try:
    inject(Scope(), scan(application))
except Exception as e:
    print(injection_trace(e))

Configurable dependencies

FunDI supports configurable dependencies - functions that return dependencies with different behavior based on provided arguments to them:

from collections.abc import Mapping

from fundi import from_, scan, inject, configurable_dependency, Scope


def require_user() -> Mapping[str, str | tuple[str]]:
    return {"username": "Kuyugama", "permissions": ("catch-apple",)}


@configurable_dependency
def require_permission(permission: str):
    def checker(user: Mapping[str, str | tuple[str]] = from_(require_user)) -> None:
        if permission not in user["permissions"]:
            raise PermissionError(permission)

    return checker


def application(
    _=from_(require_permission("catch-apple")),
):
    print("User has permission")


inject(Scope(), scan(application))

Note: configurable_dependency decorator is optional, but it caches dependencies, so their results can be cached on injection.

Also, configurable_dependency decorator does not cache dependencies configured with mutable arguments.

To get configuration of already scanned(using fundi.scan.scan) dependency - you can use CallableInfo.configuration attribute

If dependency is not scanned - use is_configured(call) function to check whether dependency is configured:

from fundi import is_configured, configurable_dependency


@configurable_dependency
def auth(optional: bool = False):
    return lambda: optional


assert is_configured(auth())

And to get dependency configuration use get_configuration(call) function on dependency callable:

import inspect

from fundi import get_configuration, configurable_dependency


@configurable_dependency
def auth(optional: bool = False):
    return lambda: optional


config = get_configuration(auth())

origin = inspect.unwrap(auth)

assert config.configurator.call == origin
assert config.values == {"optional": False}

Composite dependencies

Composite dependencies - special kind of configurable dependency that accepts other dependencies as parameters

import typing
from collections.abc import Mapping

from fundi import from_, scan, inject, configurable_dependency, Scope


def require_user() -> dict[str, str | tuple[str]]:
    return {"username": "Kuyugama", "permissions": ("catch-apple",)}


@configurable_dependency
def require_permission(
    permission: str,
    user_resolver: typing.Callable[..., Mapping[str, str | tuple[str]]] = require_user,
):
    def checker(user: Mapping[str, str | tuple[str]] = from_(user_resolver)) -> None:
        if permission not in user["permissions"]:
            raise PermissionError(permission)

    return checker


def application(
    _=from_(require_permission("catch-apple")),
):
    print("User has permission")


inject(Scope(), scan(application))

Property overriding

In some cases you will need to override dependency properties to inform FunDI about real behavior of the function.

For example, you have function that returns awaitable object, but itself is not a coroutine function. In this case you will need to override async_ property of the dependency:

from fundi import from_

def require_event_data(event: RemoteEvent):
  return api.request.event_data(event.id)

def event_handler(data: RemoteEventData = from_(require_event_data, async_=True)):
  ...

This works the same for the context and generator properties.

To define asynchronous generator or context manager use combination of async_ and generator or context.

Property inferring

FunDI makes it’s best to infer right properties using dependency’s return type-hint if it is defined.

For example, if there is function that returns context-manager and type-hint is set to contextlib.AbstractContextManager FunDI will correctly infer it as lifespan dependency.

from fundi import from_

from contextlib import AbstractContextManager

def require_session() -> AbstractContextManager[Session]:
  return session_manager.session_context()


def endpoint(session: Session = from_(require_session)): ...

This will work the same with type-hints like Generator, Awaitable or their subclasses.

Also, this will work with their asynchronous versions.