Lifespan dependency

Lifespan dependencies are dependencies with two-phase execution — preparation and tear-down.

They are usually generator-functions that yield exactly once. Before yielding it’s a preparation, and after — tear-down. The yielded result would be passed to it’s dependant.

Another way to define lifespan dependencies is using classes that implement asynchronous(__aenter__, __aexit__) or synchronous(__enter__, __exit__) context manager protocols. Their own dependencies can be defined inside the constructor(__init__). Preparation work is done in the enter method, its return value is passed to the dependant. And tear-down happens in the exit method.

Note: implicit context managers(those created using contextlib.[async]contextmanager) may not be recognized automatically. They should be explicitly marked in from_(...) or scan(...) functions using context=True parameter.

FunDI provides it own implementation of implicit context managers that are recognized as lifespan dependencies and they behave exactly as those from contextlib. Only difference is that both synchronous and asynchronous context managers are defined using a single decorator — @virtual_context

Lifespan dependencies should be used whenever tear-down logic should take the place. For example, closing a file, database session or releasing a lock.

Generator-function lifespan dependency

Example of dependency that acquires lock, opens file, yields it to dependant and closes file after it was used:

from threading import Lock

FILE_LOCK = Lock()
FILE_NAME = "file.txt"

def acquire_file():
    with FILE_LOCK:
        file = open(FILE_NAME, "w+", encoding="utf-8")
        yield file
        file.close()

I explicitly call file.close() instead of using with open(...) to make the example:

• More readable for Python beginners

• Avoid nested context managers

• Clearly show when cleanup happens

Asynchronous dependency that does the same:

from asyncio import Lock

FILE_LOCK = Lock()
FILE_NAME = "file.txt"

async def acquire_file():
    async with FILE_LOCK:
        file = open(FILE_NAME, "w+", encoding="utf-8")
        yield file
        file.close()

Context-manager lifespan dependency

You can define lifespan dependencies using class-based context managers — either synchronous (__enter__ / __exit__) or asynchronous (__aenter__ / __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))

If you want to use a function as a context manager — instead of writing a class — you can use a “virtual” context manager:

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))

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")

When to use lifespan dependencies

• Managing connections

• Working with files that need cleanup

• Acquiring and releasing locks

• Wrapping external APIs that require setup/teardown