Reference

fastapi-injectable

class fastapi_injectable.InjectableScope(*, use_cache=True)[source]

An isolated dependency lifecycle: a private exit stack + dependency cache.

Entering the scope (async with) registers it as the active scope for the current asyncio context. Any dependency resolution that happens while it is active routes its generator cleanup into this scope’s exit stack and its cached values into this scope’s cache. Leaving the scope closes the exit stack (running all cleanup) and drops the cache.

Parameters:

use_cache (bool)

property exit_stack: AsyncExitStack

The scope’s exit stack.

Push your own cleanup callbacks here to ride the same lifecycle as the scope’s dependencies.

get_cache()[source]

The scope’s dependency cache (None when caching is disabled).

Return type:

dict[Any, Any] | None

async fastapi_injectable.async_get_injected_obj(func, args=None, kwargs=None, *, use_cache=True, scope=None)[source]

Async version of get_injected_obj() for use in running event loops.

Use this function when you need to inject dependencies from within async contexts like Kafka consumers, async callbacks, or other scenarios where an event loop is already running.

This function only accepts async functions (coroutines) and async generators. For sync functions, use get_injected_obj() instead.

Parameters:

  • func (Callable[[~P], Awaitable[T]] | Callable[[~P], AsyncGenerator[T, Any]]) – The async dependency function to inject. Must be: - An async function (coroutine) - An async generator

  • args (list[Any] | None) – Positional arguments to pass to the dependency function.

  • kwargs (dict[str, Any] | None) – Keyword arguments to pass to the dependency function.

  • use_cache (bool) – Whether to cache resolved dependencies. Defaults to True.

  • scope (InjectableScope | None) – An explicit InjectableScope to route this resolution into. The scope is applied to this call’s dependency resolution within the current event loop task (its lifecycle is owned by the caller — this does not enter or close the scope). The ContextVar is set for the duration of the awaited call and reset in a finally block so it never leaks to the caller. Defaults to None (use the active scope or global manager).

Returns:

The first value yielded/returned by the dependency function after injection.

Return type:

T

Examples

```python # In an async callback (e.g., Kafka consumer) async def get_service() -> Service:

return Service()

async def consume(message):

service = await async_get_injected_obj(get_service) await service.process(message)

# With an async generator (for cleanup) async def get_db() -> AsyncGenerator[Database, None]:

db = Database() yield db await db.close()

db = await async_get_injected_obj(get_db) ```

Notes

  • This function must be called from an async context (use await)

  • Only accepts async functions and async generators

  • For sync functions, use get_injected_obj() instead

  • For async generators, only the first yielded value is returned

  • Cleanup code in generators will be executed when calling cleanup functions

  • Uses FastAPI’s dependency injection system under the hood

  • Unlike get_injected_obj(), this works in already-running event loops

async fastapi_injectable.cleanup_all_exit_stacks(*, raise_exception=False)[source]

Clean up all active exit stacks.

Parameters:

raise_exception (bool) – Whether to raise exceptions during cleanup. If False, exceptions are logged as warnings. Defaults to False.

Return type:

None

Notes

  • This method iterates through all registered exit stacks and ensures they are properly closed.

  • Typically used during application shutdown to release all managed resources.

Raises:

DependencyCleanupError – When cleanup fails and raise_exception is True

Parameters:

raise_exception (bool)

Return type:

None

async fastapi_injectable.cleanup_exit_stack_of_func(func, *, raise_exception=False)[source]

Clean up the exit stack associated with a specific function.

Parameters:

  • func (Callable[[...], Any]) – The function whose exit stack should be cleaned up.

  • raise_exception (bool) – Whether to raise exceptions during cleanup. If False, exceptions are logged as warnings. Defaults to False.

Return type:

None

Notes

  • This ensures that resources such as context managers or other async cleanup routines are properly closed for the given function.

Raises:

DependencyCleanupError – When cleanup fails and raise_exception is True

Parameters:
Return type:

None

async fastapi_injectable.clear_dependency_cache()[source]

Clear the dependency resolution cache.

Notes

  • This is useful to free up memory or reset state in scenarios where dependencies might have changed dynamically.

Return type:

None

fastapi_injectable.get_injected_obj(func, args=None, kwargs=None, *, use_cache=True, scope=None)[source]

Get an injected object from a dependency function with FastAPI’s dependency injection.

This function handles different types of callables (sync/async functions and generators) and returns the first yielded/returned value after resolving dependencies.

Parameters:

  • func (Callable[[~P], T] | Callable[[~P], Awaitable[T]] | Callable[[~P], Generator[T, Any, Any]] | Callable[[~P], AsyncGenerator[T, Any]]) – The dependency function to inject. Can be: - A regular synchronous function - An async function (coroutine) - A synchronous generator - An async generator

  • args (list[Any] | None) – Positional arguments to pass to the dependency function.

  • kwargs (dict[str, Any] | None) – Keyword arguments to pass to the dependency function.

  • use_cache (bool) – Whether to cache resolved dependencies. Defaults to True.

  • scope (InjectableScope | None) – An explicit InjectableScope to route this resolution into, without entering it as the active scope. When provided, the ContextVar is temporarily set for the duration of the call and reset in a finally block, so it never leaks to the caller. Defaults to None (use the active scope or global manager).

Returns:

The first value yielded/returned by the dependency function after injection.

Return type:

T

Examples

```python # With a regular function def get_service() -> Service:

return Service()

service = get_injected_obj(get_service)

# With an async function async def get_async_service() -> Service:

return await create_service()

service = get_injected_obj(get_async_service)

# With a generator (for cleanup) def get_db() -> Generator[Database, None, None]:

db = Database() yield db db.cleanup()

db = get_injected_obj(get_db) ```

Notes

  • For generator functions, only the first yielded value is returned

  • Cleanup code in generators will be executed when calling cleanup functions

  • Uses FastAPI’s dependency injection system under the hood

  • The scope parameter is async-first. Under the background_thread loop strategy the ContextVar is not propagated across threads, so an explicit scope is not visible inside the background loop. Prefer the async API (async_get_injected_obj / injectable_scope) for scoped resolution.

fastapi_injectable.injectable(func=None, *, use_cache=True)[source]

Decorator to inject dependencies into any callable, sync or async.

Parameters:
Return type:

Callable[[…], Any]

fastapi_injectable.injectable_scope(*, use_cache=True)[source]

Open an isolated dependency scope for the duration of the async with block.

Inside the block every dependency resolved (via async_get_injected_obj or by calling an @injectable function) uses this scope’s private exit stack and cache. Leaving the block cleans up only this scope’s resources, isolated from any other concurrent scope.

Example:

async def process_event(event):
    async with injectable_scope():
        dep = await async_get_injected_obj(get_dep)
        await handle(event, dep)
    # only this event's resources are cleaned up here
Parameters:

use_cache (bool)

Return type:

AsyncIterator[InjectableScope]

async fastapi_injectable.register_app(app)[source]

Register the given FastAPI app for constructing fake request later.

Parameters:

app (FastAPI)

Return type:

None

async fastapi_injectable.resolve_dependencies(func, *, use_cache=True, provided_kwargs=None)[source]

Resolve dependencies for the given function using FastAPI’s dependency injection system.

This function resolves dependencies defined via FastAPI’s dependency mechanism and returns a dictionary of resolved arguments for the given function.

Parameters:

  • func (Callable[[~P], T] | Callable[[~P], Awaitable[T]]) – The function for which dependencies need to be resolved. It can be a synchronous or asynchronous callable.

  • use_cache (bool) – Whether to use a cache for dependency resolution. Defaults to True.

  • provided_kwargs (dict[str, Any] | None) – Explicit kwargs passed by the caller (these override DI).

Returns:

A dictionary mapping argument names to resolved dependency values.

Return type:

dict[str, Any]

Notes

  • A fake HTTP request is created to mimic FastAPI’s request-based dependency resolution.

fastapi_injectable.run_coroutine_sync(coro)[source]

Run an async coroutine synchronously.

Parameters:

coro (Coroutine[Any, Any, T]) – The coroutine to execute.

Returns:

The result of the coroutine execution.

Raises:

Any exception raised by the coroutine or during execution. –

Return type:

T

fastapi_injectable.setup_graceful_shutdown(signals=None, *, raise_exception=False)[source]

Register handlers to perform cleanup during application shutdown.

Parameters:

  • signals (Sequence[Signals] | None) – A list of OS signals that should trigger the cleanup process. Defaults to [SIGINT, SIGTERM].

  • raise_exception (bool) – Whether to raise exceptions during cleanup. If False, exceptions are logged as warnings. Defaults to False.

Return type:

None

Notes

  • When a registered signal is received, this function ensures that all resources (e.g., exit stacks) are properly released before the application exits.

  • Also registers a cleanup routine via atexit to handle unexpected shutdown scenarios.

Raises:

DependencyCleanupError – When cleanup fails and raise_exception is True

Parameters:
Return type:

None