DOCSEARCHbyAlgolia
Submit Docs

fastapi-injectable

repository·main·Indexed 43 days ago

https://github.com/jaspersui/fastapi-injectable

A lightweight package that enables the use of FastAPI's dependency injection system (Depends()) outside of route handlers. It allows developers to reuse dependency logic in CLI tools, Celery tasks, background workers, and scheduled jobs without duplicating code. Version 1.6.1 supports Python 3.10+ and FastAPI >=0.112.4.

Tokens
10.1K
Snippets
35
Records
47
Agent score
54%

What's inside fastapi-injectable

  1. Key features of fastapi-injectable

    main

    Core Capabilities

    • Drop-in decorator: Add @injectable to any function using Depends().
    • Full async support: Works with sync, async, and mixed dependency chains.
    • Test-friendly: Manual overrides let you swap in mocks instantly.
    • Resource cleanup: Built-in lifecycle management for generator dependencies.
    • Dependency caching: Optional caching for better performance.
    • App state access: Register your FastAPI app to access app.state in dependencies.
    • Mypy plugin: Opt-in plugin for full type-checking.
    • Graceful shutdown: Automatic cleanup on program exit via signal handling.

  2. Integrate @injectable with task and worker frameworks

    main

    You can use @injectable with most task/worker frameworks (like Celery, Dramatiq, or Temporal) by stacking decorators.

    The Rule: Always place the framework's decorator on the outside and @injectable on the inside (closest to the function definition).

    When the framework calls the function, @injectable intercepts the call to resolve any Depends(...) parameters. The caller provides the standard arguments, and @injectable provides the dependencies. If the caller provides keyword arguments that match dependency names, those caller-supplied values will override the injected ones.

    @framework_decorator  # outermost
@injectable           # innermost
def my_task(arg1: int, dep: Annotated[Type, Depends(get_dep)]):
    ...

  3. Use `injectable_scope()` for isolated dependency scopes

    main

    By default, fastapi-injectable uses a process-global, function-keyed cache and exit stack. This means concurrent calls to the same function share the same resources, which is unsafe for parallel tasks (e.g., using asyncio.TaskGroup).

    Use injectable_scope() to create a request-scoped model similar to FastAPI's internal behavior. Each scope provides its own isolated exit stack and cache, ensuring that cleaning up one task's resources does not affect others. This is the recommended approach for async consumers fanning out work.

    import asyncio
from collections.abc import AsyncGenerator
from fastapi_injectable import async_get_injected_obj, injectable_scope

async def get_connection() -> AsyncGenerator[Connection, None]:
    conn = await open_connection()
    try:
        yield conn
    finally:
        await conn.close()

async def process_event(event) -> None:
    # Each task gets its own isolated stack and cache
    async with injectable_scope():
        conn = await async_get_injected_obj(get_connection)
        await handle(event, conn)
    # only this event's connection is closed here — siblings are untouched

async def consume(events) -> None:
    async with asyncio.TaskGroup() as tg:
        for event in events:
            tg.create_task(process_event(event))

  4. Configure Event Loop strategies via loop_manager

    main

    Use loop_manager.set_loop_strategy() to control how asynchronous code is executed. This is critical when running async code in sync environments or managing objects like aiohttp.ClientSession that are loop-bound.

    Available strategies:

    1. current (default): Uses the current thread's event loop. Fails if no loop is running.
    2. isolated: Creates a separate, dedicated event loop. Ideal for scripts or when you need to ensure objects are created in a specific loop.
    3. background_thread: Runs a dedicated event loop in a background daemon thread. Best for long-running applications where you need to run async code from sync contexts without blocking.
    from fastapi_injectable import loop_manager

# Set to isolated to ensure a dedicated loop for injected objects
loop_manager.set_loop_strategy("isolated")
loop = loop_manager.get_loop()

  5. Manage dependency cleanup in long-running processes

    main

    Because dependencies are cached in a process-global cache by default, long-running workers must manage state to prevent memory leaks or stale resources. You have several options:

    1. Per-function cleanup: Use await cleanup_exit_stack_of_func(your_func).
    2. Global cleanup: Use await cleanup_all_exit_stacks().
    3. Cache reset: Use await clear_dependency_cache() to clear the process-global cache.
    4. Graceful shutdown: Use setup_graceful_shutdown() to handle cleanup on program termination (SIGTERM/SIGINT).
    5. Scoped approach (Recommended): Wrap each unit of work in an injectable_scope() block, which handles both resource cleanup and cache isolation automatically.

  6. Resolve dependencies manually in CLI frameworks (Typer/Click)

    main

    CLI frameworks like Typer or Click inspect function signatures to generate command-line options. If you use @injectable on a CLI command, the injected dependencies will incorrectly appear as CLI options.

    To avoid this, do not use the @injectable decorator on CLI commands. Instead, resolve dependencies manually inside the function body using get_injected_obj(dependency_function).

    Important: Since CLI commands are often short-lived processes, ensure you clean up resources (like database connections or caches) before the process exits using run_coroutine_sync(cleanup_all_exit_stacks()) and run_coroutine_sync(clear_dependency_cache()).

    import typer
from fastapi_injectable import (
    cleanup_all_exit_stacks,
    clear_dependency_cache,
    get_injected_obj,
    run_coroutine_sync,
)

cli = typer.Typer()

@cli.command()
def migrate(target: str) -> None:
    # Resolve inside the body to prevent leaking into CLI options
    db: DbSession = get_injected_obj(get_db) 
    typer.echo(f"migrating to {target} (connected={db.connected})")

    # Clean up before the process exits
    run_coroutine_sync(cleanup_all_exit_stacks())
    run_coroutine_sync(clear_dependency_cache())

if __name__ == "__main__":
    cli()

  7. Override dependencies manually

    main

    You can override dependencies by passing arguments directly to an @injectable function. Any arguments passed explicitly will take priority over the values resolved via FastAPI's dependency injection system. This is highly useful for unit testing with mocks.

    @injectable
def process_data(db: Annotated[Database, Depends(get_db)]) -> str:
    return db.query()

# Override dependency manually (great for tests)
mock_db = Database()
mock_db.query = lambda: "mock data"

print(process_data(db=mock_db)) # Explicitly pass the mock dependency
# Output: "mock data"

  9. Register FastAPI app to access app.state in dependencies

    main

    If your dependencies need to access shared services stored in app.state (like a database engine), you must register the FastAPI app using await register_app(app). This is typically done within the FastAPI lifespan handler.

    @contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    # Register the app so dependencies can access app.state
    await register_app(app)
    app.state.db_engine = create_async_engine("...")
    yield
    await app.state.db_engine.dispose()

app = FastAPI(lifespan=lifespan)

  10. Configure mypy plugin for fastapi-injectable

    main

    To ensure @injectable decorated functions are type-checked correctly by mypy, add the bundled plugin to your mypy configuration file.

    Note: This plugin is for mypy only. pyright/Pylance users may still see call-arg errors.

    [tool.mypy]
plugins = ["fastapi_injectable.mypy"]

  11. Test nested dependencies using app.dependency_overrides

    main

    Manual argument overrides only work for top-level parameters of the entry-point function. To override a nested dependency deeper in the chain, use FastAPI's native app.dependency_overrides. For this to work, you must first call register_app(app) so fastapi-injectable can consult the FastAPI app's overrides.

    from fastapi import FastAPI
from fastapi_injectable import register_app, run_coroutine_sync

app = FastAPI()
# Register the app so fastapi-injectable consults app.dependency_overrides.
run_coroutine_sync(register_app(app))

# Override a nested dependency
app.dependency_overrides[get_db] = lambda: Database("mock")

try:
    assert handle_request() == "mock"
finally:
    app.dependency_overrides.clear()

  12. Configure mypy support for @injectable

    main

    Static type checkers like mypy will report errors (e.g., call-arg) when calling @injectable functions without arguments. To resolve this, enable the fastapi_injectable.mypy plugin in your configuration file.

    For pyproject.toml:

    [tool.mypy]
plugins = ["fastapi_injectable.mypy"]

    Note: This is a mypy plugin only. pyright/Pylance will still report errors.