fastapi-depends-anywhere

Run FastAPI dependency injection anywhere - in background tasks, scripts, tests, and migrations.

Installation

# Using pip
pip install fastapi-depends-anywhere

# Using uv
uv add fastapi-depends-anywhere

# With asyncer support (for runnify_with_fastapi_depends)
pip install fastapi-depends-anywhere[asyncer]

Quick Start

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi_depends_anywhere import configure, with_fastapi_depends

app = FastAPI()

# Configure once at startup
configure(app=app)

async def get_db() -> Database:
    """Your database dependency."""
    return Database()

DbDep = Annotated[Database, Depends(get_db)]

@with_fastapi_depends
async def background_task(*, db: DbDep) -> None:
    """Background task that uses FastAPI dependencies."""
    await db.execute("INSERT INTO logs ...")

# Run it - dependencies are automatically resolved!
await background_task()

Why This Library?

FastAPI's dependency injection is powerful, but it only works inside route handlers. This library lets you use the same dependencies in:

Background tasks - Process jobs with full dependency access
Admin scripts - Run one-off scripts with database connections
Tests - Test functions that use dependencies without mocking everything
Migrations - Run database migrations with proper dependency injection
CLI commands - Build CLI tools that reuse your FastAPI dependencies

Core Features

`with_fastapi_depends`

Decorator that resolves FastAPI dependencies for any async or sync function:

from fastapi_depends_anywhere import with_fastapi_depends

@with_fastapi_depends
async def send_notification(user_id: int, *, db: DbDep, mailer: MailerDep) -> None:
    user = await db.get_user(user_id)
    await mailer.send(user.email, "Hello!")

# Dependencies are resolved automatically
await send_notification(123)

`aiter_with_fastapi_depends`

Decorator for async generators:

from fastapi_depends_anywhere import aiter_with_fastapi_depends

@aiter_with_fastapi_depends
async def stream_users(*, db: DbDep) -> AsyncGenerator[User, None]:
    async for user in db.stream_all_users():
        yield user

async for user in stream_users():
    print(user.name)

`with_fastapi_lifecycle`

Runs your function within FastAPI's lifespan context (startup/shutdown) when you're outside the normal request flow - such as in standalone scripts, CLI tools, or background workers:

from fastapi_depends_anywhere import with_fastapi_lifecycle

@with_fastapi_lifecycle
async def run_migration() -> None:
    # FastAPI's startup has run - resources are initialized!
    # (database connections, caches, etc.)
    async with get_db() as db:
        await run_all_migrations(db)
    # FastAPI's shutdown will run after this - cleanup happens automatically

# In a standalone script
asyncio.run(run_migration())

This is essential when your dependencies rely on resources initialized during FastAPI's lifespan (e.g., database connection pools, Redis clients, ML models).

`runnify_with_fastapi_depends`

For CLI scripts - combines lifecycle, dependency resolution, and sync execution:

from fastapi_depends_anywhere import runnify_with_fastapi_depends

@runnify_with_fastapi_depends
async def cli_report(*, db: DbDep) -> None:
    results = await db.fetch_all("SELECT * FROM reports")
    for row in results:
        print(row)

# No asyncio.run() needed!
if __name__ == "__main__":
    cli_report()

Requires the asyncer extra: pip install fastapi-depends-anywhere[asyncer]

`FastApiDepsStack`

A stateful context manager for resolving multiple dependencies in long-running, non-server contexts such as notebooks, interactive scripts, and workers. Starts the FastAPI lifespan once, keeps all resolved instances alive across calls, and tears everything down on close.

from fastapi_depends_anywhere import FastApiDepsStack

async with FastApiDepsStack(app=app) as stack:
    db = await stack.resolve(DbDep)
    cache = await stack.resolve(CacheDep)
    # db and cache stay alive for the duration of the block

Or with explicit start/close:

stack = FastApiDepsStack(app=app)
await stack.start()

db = await stack.resolve(DbDep)
cache = await stack.resolve(CacheDep)

await stack.close()  # runs all dep teardown + lifespan shutdown

Synchronous Execution

Use the sync subpackage to call FastAPI-dependency-injected functions from a synchronous context — Celery tasks, Django views, CLI scripts — without managing an event loop yourself.

All sync decorators accept both sync and async handler functions and always return a plain sync callable.

from fastapi_depends_anywhere.sync import with_fastapi_depends, with_fastapi_lifecycle

@with_fastapi_depends
def process_record(record_id: int, *, db: DbDep) -> Result:
    return db.fetch(record_id)

# No await, no asyncio.run() — just call it
result = process_record(42)

For scripts that also need the FastAPI lifespan (startup/shutdown):

from fastapi_depends_anywhere.sync.runners import runnify_with_fastapi_depends

@runnify_with_fastapi_depends
def nightly_cleanup(*, db: DbDep) -> None:
    db.delete_old_records(days=30)

if __name__ == "__main__":
    nightly_cleanup()  # lifecycle + deps fully managed

For async generators bridged to sync iteration:

from fastapi_depends_anywhere.sync import iter_with_fastapi_depends

@iter_with_fastapi_depends
def stream_users(*, db: DbDep) -> Generator[User, None, None]:
    yield from db.iter_all_users()

for user in stream_users():
    print(user.name)

The sync FastApiDepsStack works the same way:

from fastapi_depends_anywhere.sync import FastApiDepsStack

with FastApiDepsStack(app=app) as stack:
    db = stack.resolve(DbDep)
    cache = stack.resolve(CacheDep)

How it works: a single daemon thread hosts a persistent asyncio event loop shared across all sync calls. Dependency resolution runs there via run_coroutine_threadsafe, so there is no per-call event loop creation overhead. No extra dependencies required.

Configuration

Global Configuration

Configure once at application startup:

from fastapi import FastAPI
from fastapi_depends_anywhere import configure

app = FastAPI()

# Basic configuration
configure(app=app)

# With context logging integration
from context_logging import Context
configure(app=app, context_factory=lambda ctx: Context(**ctx))

Explicit App Parameter

Or pass the app explicitly to each decorator:

@with_fastapi_depends(app=my_app)
async def my_function(*, db: DbDep) -> None:
    ...

Advanced Usage

Dependency Overrides

Dependency overrides work just like in FastAPI:

app.dependency_overrides[get_db] = get_test_db

@with_fastapi_depends
async def my_function(*, db: DbDep) -> None:
    # Uses get_test_db instead of get_db
    ...

Custom Request Scope

Pass custom ASGI scope data to dependencies that need it.

At decoration time (static scope):

@with_fastapi_depends(scope={"method": "POST", "path": "/custom"})
async def my_function(*, request: Request) -> None:
    print(request.method)  # "POST"

At call time (dynamic scope) - useful for background tasks that need request context:

@with_fastapi_depends
async def process_for_user(*, user: CurrentUserDep) -> None:
    # user is resolved from the passed scope's headers
    print(f"Processing for {user.id}")

@app.post("/trigger")
async def trigger(request: Request, background_tasks: BackgroundTasks) -> dict:
    # Pass the request scope to preserve auth headers, etc.
    background_tasks.add_task(process_for_user, _scope=request.scope)
    return {"status": "queued"}

This allows dependencies that read from Request (like auth) to work in background tasks:

from fastapi import Header

async def get_current_user(authorization: str = Header()) -> AuthUser:
    if authorization.startswith("Bearer "):
        return decode_token(authorization[7:])
    raise HTTPException(401)

CurrentUserDep = Annotated[AuthUser, Depends(get_current_user)]

@with_fastapi_depends
async def send_notification(*, user: CurrentUserDep, mailer: MailerDep) -> None:
    await mailer.send(user.email, "Task completed!")

# In your route - scope carries the auth headers
background_tasks.add_task(send_notification, _scope=request.scope)

Context Integration

For libraries like context_logging:

from context_logging import Context

configure(
    app=app,
    context_factory=lambda ctx: Context(**ctx)
)

@with_fastapi_depends(context={"request_id": "abc123"})
async def my_function(*, db: DbDep) -> None:
    # Context variables are set
    ...

Validation Utilities

Check if all routes and dependencies are async (recommended for context variable propagation):

from fastapi_depends_anywhere import check_routes_and_dependencies_are_async

@app.on_event("startup")
async def startup() -> None:
    check_routes_and_dependencies_are_async(app)

API Reference

Configuration

configure(app=None, context_factory=None) - Set global configuration
get_app() - Get the configured FastAPI app
reset_config() - Reset configuration (useful for tests)

Decorators (async)

with_fastapi_depends(func, scope=None, context=None, app=None) - Resolve dependencies for a function
aiter_with_fastapi_depends(func, app=None) - Resolve dependencies for an async generator
with_fastapi_lifecycle(func, app=None) - Run within FastAPI lifespan (for scripts/workers outside request flow)
runnify_with_fastapi_depends(func, app=None) - Run async function synchronously with dependencies (requires asyncer)

Dependency Stack (async)

FastApiDepsStack(app=None) - stateful async context manager; call start(), resolve(DepType), close() (or use async with)

Context Manager (async)

resolve_fastapi_depends(func, scope=None, dependency_overrides_provider=None) - Low-level async context manager for dependency resolution