Skip to content

Lifespan

The lifespan module provides composable FastAPI lifespan management with type-safe dependency injection.

Overview

FastAPI accepts only one lifespan callable, but applications often need to manage multiple independent resources—database pools, cache clients, HTTP sessions—each with its own setup and teardown logic. The Lifespan class composes any number of async context managers (hooks) into a single lifespan, storing each hook's yielded resource so that dependency injection functions can retrieve it with full type preservation.

Basic Usage

import contextlib
import typing
import fastapi
from collections import abc
from imbi_common import lifespan

type PoolType = ...  # your connection pool type


@contextlib.asynccontextmanager
async def db_hook() -> abc.AsyncIterator[PoolType]:
    pool = PoolType(...)
    await pool.open()
    try:
        yield pool
    finally:
        await pool.close()


async def _inject_pool(
    context: lifespan.InjectLifespan,
) -> PoolType:
    return context.get_state(db_hook)


DbPool = typing.Annotated[PoolType, fastapi.Depends(_inject_pool)]

app = fastapi.FastAPI(lifespan=lifespan.Lifespan(db_hook))


@app.get('/items')
async def list_items(*, pool: DbPool) -> list[dict]:
    ...

See the Lifespan and Dependency Injection guide for a full walkthrough of the pattern.

API Reference

Type Aliases

LifespanHook 

LifespanHook = Callable[
    [], AbstractAsyncContextManager[object | None]
]

TypedLifespanHook 

TypedLifespanHook = Callable[
    [], AbstractAsyncContextManager[T | None]
]

InjectLifespan 

InjectLifespan = Annotated[Lifespan, Depends(_get_lifespan)]

Classes

Lifespan 

Lifespan(*hooks: LifespanHook)

Bases: dict[LifespanHook, object | None]

Compose multiple lifespan hooks into a single FastAPI lifespan.

Manages multiple independent async context managers (lifespan hooks) and provides type-safe access to their yielded resources through dependency injection. Hooks are deduplicated (same hook only runs once) and cleaned up in LIFO order.

Parameters:

Name Type Description Default
*hooks LifespanHook

Variable number of lifespan hooks to combine. Each hook is an async context manager that yields a resource.

 ()
Example

::

@contextlib.asynccontextmanager
async def postgres_lifespan() -> abc.AsyncIterator[PoolType]:
    async with psycopg_pool.AsyncConnectionPool(...) as pool:
        yield pool

app = fastapi.FastAPI(
    lifespan=Lifespan(postgres_lifespan, redis_lifespan)
)
See Also

get_state: Retrieve resources from hooks with type preservation InjectLifespan: Type alias for dependency injection

Initialize Lifespan with the given hooks.

Parameters:

Name Type Description Default
*hooks LifespanHook

Variable number of lifespan hooks to combine. Hooks are entered in the order provided and exited in LIFO order. Duplicate hooks are deduplicated automatically.

 ()
Source code in src/imbi_common/lifespan.py 
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(self, *hooks: LifespanHook) -> None:
    """Initialize Lifespan with the given hooks.

    Args:
        *hooks (LifespanHook): Variable number of lifespan hooks to
            combine. Hooks are entered in the order provided and
            exited in LIFO order. Duplicate hooks are deduplicated
            automatically.
    """
    super().__init__()
    self._hooks: tuple[LifespanHook, ...] = hooks

__call__ 

__call__(
    _app: FastAPI,
) -> contextlib.AbstractAsyncContextManager[
    dict[str, Lifespan]
]

Make Lifespan callable as a FastAPI lifespan function.

This method is called automatically by FastAPI during application startup. It enters all registered hooks, stores their yielded resources, and ensures proper cleanup on shutdown.

Parameters:

Name Type Description Default
_app FastAPI

The FastAPI application instance (unused, required by FastAPI lifespan protocol).

 required

Returns:

Type Description
AbstractAsyncContextManager[dict[str, Lifespan]]

contextlib.AbstractAsyncContextManager[dict[str, Lifespan]]: An async context manager that yields a dictionary containing the Lifespan instance under the key 'lifespan_data'.
Note
  • Hooks are entered in the order provided to init
  • Duplicate hooks are detected and only executed once
  • Resources are cleaned up in LIFO order (last-in-first-out)
  • Uses AsyncExitStack to ensure proper cleanup even if hooks raise exceptions
Source code in src/imbi_common/lifespan.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def __call__(
    self, _app: fastapi.FastAPI
) -> contextlib.AbstractAsyncContextManager[dict[str, 'Lifespan']]:
    """Make Lifespan callable as a FastAPI lifespan function.

    This method is called automatically by FastAPI during application
    startup. It enters all registered hooks, stores their yielded
    resources, and ensures proper cleanup on shutdown.

    Args:
        _app (fastapi.FastAPI): The FastAPI application instance
            (unused, required by FastAPI lifespan protocol).

    Returns:
        contextlib.AbstractAsyncContextManager[dict[str, Lifespan]]:
            An async context manager that yields a dictionary
            containing the Lifespan instance under the key
            'lifespan_data'.

    Note:
        - Hooks are entered in the order provided to __init__
        - Duplicate hooks are detected and only executed once
        - Resources are cleaned up in LIFO order (last-in-first-out)
        - Uses AsyncExitStack to ensure proper cleanup even if hooks
          raise exceptions
    """

    @contextlib.asynccontextmanager
    async def cm() -> abc.AsyncIterator[dict[str, 'Lifespan']]:
        async with contextlib.AsyncExitStack() as stack:
            for hook in self._hooks:
                if hook not in self:
                    self[hook] = await stack.enter_async_context(hook())
            yield {'lifespan_data': self}

    return cm()

get_state 

get_state(hook: TypedLifespanHook[T]) -> T

Retrieve the resource yielded by a specific hook.

This is a generic method that preserves type information. If the hook yields a resource of type T, this method returns T.

Parameters:

Name Type Description Default
hook TypedLifespanHook[T]

The lifespan hook whose resource to retrieve. Must have been passed to the Lifespan constructor.

 required

Returns:

Name Type Description
T T

The resource yielded by the hook, with type preserved.

Raises:

Type Description
HTTPException

500 error if the hook was not registered with this Lifespan instance.
Example

::

def _inject_pool(context: InjectLifespan) -> PoolType:
    # Type of pool is PoolType (not object)
    pool = context.get_state(postgres_lifespan)
    return pool
Source code in src/imbi_common/lifespan.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def get_state[T](self, hook: TypedLifespanHook[T]) -> T:
    """Retrieve the resource yielded by a specific hook.

    This is a generic method that preserves type information. If the
    hook yields a resource of type `T`, this method returns `T`.

    Args:
        hook (TypedLifespanHook[T]): The lifespan hook whose
            resource to retrieve. Must have been passed to the
            Lifespan constructor.

    Returns:
        T: The resource yielded by the hook, with type preserved.

    Raises:
        fastapi.HTTPException: 500 error if the hook was not
            registered with this Lifespan instance.

    Example:
        ::

            def _inject_pool(context: InjectLifespan) -> PoolType:
                # Type of pool is PoolType (not object)
                pool = context.get_state(postgres_lifespan)
                return pool
    """
    try:
        return t.cast('T', self[hook])
    except KeyError:
        raise fastapi.HTTPException(
            http.HTTPStatus.INTERNAL_SERVER_ERROR,
            detail=f'Unmet lifespan dependency hook {hook!r}',
        ) from None