Basic Usage¶

fastapi-injectable provides several powerful ways to use FastAPI’s dependency injection outside of route handlers. Let’s explore the key usage patterns with practical examples.

Basic Injection¶

The most basic way to use dependency injection is through the @injectable decorator. This allows you to use FastAPI’s Depends in any function, not just route handlers.

from typing import Annotated

from fastapi import Depends
from fastapi_injectable.decorator import injectable

class Database:
    def __init__(self) -> None:
        pass

    def query(self) -> str:
        return "data"

# Define your dependencies
def get_database():
    return Database()

# Use dependencies in any function
@injectable
def process_data(db: Annotated[Database, Depends(get_database)]):
    return db.query()

# Call it like a normal function
result = process_data()
print(result) # Output: 'data'

Function-based Approach¶

The function-based approach provides an alternative way to use dependency injection without decorators. This can be useful when you need more flexibility or want to avoid modifying the original function.

Here’s how to use it:

from fastapi_injectable.util import get_injected_obj

class Database:
    def __init__(self) -> None:
        pass

    def query(self) -> str:
        return "data"

def process_data(db: Annotated[Database, Depends(get_database)]):
    return db.query()

# Get injected instance without decorator
result = get_injected_obj(process_data)
print(result) # Output: 'data'

Generator Dependencies with Cleanup¶

When working with generator dependencies that require cleanup (like database connections or file handles), fastapi-injectable provides built-in support for controlling dependency lifecycles and proper resource management with error handling.

Here’s an example showing how to work with generator dependencies:

from collections.abc import Generator

from fastapi_injectable.util import cleanup_all_exit_stacks, cleanup_exit_stack_of_func
from fastapi_injectable.exception import DependencyCleanupError

class Database:
    def __init__(self) -> None:
        self.closed = False

    def query(self) -> str:
        return "data"

    def close(self) -> None:
        self.closed = True

class Machine:
    def __init__(self, db: Database) -> None:
        self.db = db

def get_database() -> Generator[Database, None, None]:
    db = Database()
    yield db
    db.close()

@injectable
def get_machine(db: Annotated[Database, Depends(get_database)]):
    machine = Machine(db)
    return machine

# Use the function
machine = get_machine()

# Option #1: Silent cleanup when done for a single decorated function (logs errors but doesn't raise)
assert machine.db.closed is False
await cleanup_exit_stack_of_func(get_machine)
assert machine.db.closed is True

# Option #2: Strict cleanup with error handling
try:
    await cleanup_exit_stack_of_func(get_machine, raise_exception=True)
except DependencyCleanupError as e:
    print(f"Cleanup failed: {e}")

# Option #3: If you don't care about the other injectable functions,
#              just use the cleanup_all_exit_stacks() to cleanup all at once.
assert machine.db.closed is False
await cleanup_all_exit_stacks() # can still pass the raise_exception=True to raise the error if you want
assert machine.db.closed is True

Async Support¶

fastapi-injectable provides full support for both synchronous and asynchronous dependencies, allowing you to mix and match them as needed. You can freely use async dependencies in sync functions and vice versa. For cases where you need to run async code in a synchronous context, we provide the run_coroutine_sync utility function.

from collections.abc import AsyncGenerator

class AsyncDatabase:
    def __init__(self) -> None:
        self.closed = False

    async def query(self) -> str:
        return "data"

    async def close(self) -> None:
        self.closed = True

async def get_async_database() -> AsyncGenerator[AsyncDatabase, None]:
    db = AsyncDatabase()
    yield db
    await db.close()

@injectable
async def async_process_data(db: Annotated[AsyncDatabase, Depends(get_async_database)]):
    return await db.query()

# Use it with async/await
result = await async_process_data()
print(result) # Output: 'data'

# In sync func, you can still get the result by using `run_coroutine_sync()`
from fastapi_injectable.concurrency import run_coroutine_sync

result = run_coroutine_sync(async_process_data())
print(result) # Output: 'data'

Dependency Caching Control¶

By default, fastapi-injectable caches dependency instances to improve performance and maintain consistency. This means when you request a dependency multiple times, you’ll get the same instance back.

You can control this behavior using the use_cache parameter in the @injectable decorator:

use_cache=True (default): Dependencies are cached and reused
use_cache=False: New instances are created for each dependency request

Using use_cache=False is particularly useful when:

You need fresh instances for each request
You want to avoid sharing state between different parts of your application
You’re dealing with stateful dependencies that shouldn’t be reused

from typing import Annotated

from fastapi import Depends

from fastapi_injectable.decorator import injectable

class Mayor:
    pass

class Capital:
    def __init__(self, mayor: Mayor) -> None:
        self.mayor = mayor

class Country:
    def __init__(self, capital: Capital) -> None:
        self.capital = capital

def get_mayor() -> Mayor:
    return Mayor()

def get_capital(mayor: Annotated[Mayor, Depends(get_mayor)]) -> Capital:
    return Capital(mayor)

@injectable
def get_country(capital: Annotated[Capital, Depends(get_capital)]) -> Country:
    return Country(capital)

# With caching (default), all instances share the same dependencies
country_1 = get_country()
country_2 = get_country()
country_3 = get_country()
assert country_1.capital is country_2.capital is country_3.capital
assert country_1.capital.mayor is country_2.capital.mayor is country_3.capital.mayor

# Without caching, new instances are created each time
@injectable(use_cache=False)
def get_country(capital: Annotated[Capital, Depends(get_capital)]) -> Country:
    return Country(capital)

country_1 = get_country()
country_2 = get_country()
country_3 = get_country()
assert country_1.capital is not country_2.capital is not country_3.capital
assert country_1.capital.mayor is not country_2.capital.mayor is not country_3.capital.mayor

Graceful Shutdown¶

If you want to ensure proper cleanup when the program exits, you can register cleanup functions with error handling:

import signal

from fastapi_injectable import setup_graceful_shutdown
from fastapi_injectable.exception import DependencyCleanupError

# Option #1: Silent cleanup (default)
# it handles SIGTERM and SIGINT, and will logs errors if any exceptions are raised during cleanup
setup_graceful_shutdown()

# Option #2: Strict cleanup that raises errors
# it handles SIGTERM and SIGINT, and will raise DependencyCleanupError if any exceptions are raised during cleanup
setup_graceful_shutdown(raise_exception=True)

# Option #3: Pass custom signals to handle
# it handles the custom signals, and will raise DependencyCleanupError if any exceptions are raised during cleanup
setup_graceful_shutdown(
    signals=[signal.SIGTERM],
    raise_exception=True
)

App Registration for State Access¶

If your dependencies need access to the FastAPI app state (like database connections or other services), you can register your app with fastapi-injectable:

from fastapi import FastAPI, Request, Depends
from fastapi_injectable import injectable, register_app
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession

# Define your dependencies that need app state access
def get_db_engine(*, request: Request) -> AsyncEngine:
    return request.app.state.db_engine

DBEngine = Annotated[AsyncEngine, Depends(get_db_engine)]

async def get_db(*, db_engine: DBEngine) -> AsyncIterator[AsyncSession]:
    session = async_sessionmaker(db_engine)
    async with session.begin() as session:
        yield session

DB = Annotated[AsyncSession, Depends(get_db)]

# Register your app during startup
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    # Register the app so dependencies can access app.state
    await register_app(app)

    # Setup your app state
    app.state.db_engine = create_async_engine("postgresql+asyncpg://...")
    yield
    await app.state.db_engine.dispose()

app = FastAPI(lifespan=lifespan)

# Now you can use dependencies that need app state anywhere!
@injectable
async def process_data(db: DB) -> str:
    result = await db.execute(...)
    return result

# Use it in background tasks, CLI tools, etc.
result = await process_data()

This is particularly useful when:

Your dependencies need access to shared services in app.state
You’re using third-party libraries that call your code internally
You want to maintain a single source of truth for long-running services