Use FastAPI's Depends() anywhere - even outside FastAPI routes
Installation: pip install fastapi-injectable
Documentation: https://fastapi-injectable.readthedocs.io/en/latest/
from typing import Annotated
from fastapi import Depends
from fastapi_injectable import injectable
class Database:
def query(self) -> str:
return "data"
def get_db() -> Database:
return Database()
@injectable
def process_data(db: Annotated[Database, Depends(get_db)]) -> str:
return db.query()
# Use it anywhere!
result = process_data()
print(result) # Output: 'data'- Flexible Injection: Use decorators, function wrappers, or utility functions.
- Full Async Support: Works with both sync and async code.
- Resource Management: Built-in cleanup for dependencies.
- Dependency Caching: Optional caching for better performance.
- Graceful Shutdown: Automatic cleanup on program exit.
fastapi-injectable is a lightweight package that enables seamless use of FastAPI's dependency injection system outside of route handlers. It solves a common pain point where developers need to reuse FastAPI dependencies in non-FastAPI contexts like CLI tools, background tasks, or scheduled jobs, allowing you to use FastAPI's dependency injection system anywhere!
- Python
3.10or higher - FastAPI
0.112.4or higher
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.
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'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'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 Truefastapi-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'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 reuseduse_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.mayorIf 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
)If the basic examples don't cover your needs, check out our test files - they're basically a cookbook of real-world scenarios:
1. test_injectable.py - Shows all possible combinations of:
- Sync/async functions
- Decorator vs function wrapping
- Caching vs no caching
2. test_integration.py - Demonstrates:
- Resource cleanup
- Generator dependencies
- Mixed sync/async dependencies
- Multiple dependency chains
These test cases mirror common development patterns you'll encounter. They show how to handle complex dependency trees, resource management, and mixing sync/async code - stuff you'll actually use in production.
The test files are written to be self-documenting, so browsing through them will give you practical examples for most scenarios you'll face in your codebase.
We've collected some real-world examples of using fastapi-injectable in various scenarios:
This example demonstrates several key patterns for using dependency injection in background workers:
-
Fresh Dependencies per Message:
- Each message gets a fresh set of dependencies through
_init_as_consumer() - This ensures clean state for each message, similar to how FastAPI handles HTTP requests
- Each message gets a fresh set of dependencies through
-
Proper Resource Management:
- Dependencies with cleanup needs (like database connections) are properly handled
- Cleanup code in generators runs when
cleanup_exit_stack_of_func()is called - Cache is cleared between messages to prevent memory leaks
-
Graceful Shutdown:
setup_graceful_shutdown()ensures resources are cleaned up on program termination- Handles both SIGTERM and SIGINT signals
Please refer to the Real-world Examples for more details.
- Why would I need this package?
- Why not directly use other DI packages like Dependency Injector or FastDepends?
- Can I use it with existing FastAPI dependencies?
- Does it work with all FastAPI dependency types?
- What happens to dependency cleanup in long-running processes?
- Can I mix sync and async dependencies?
- Are type hints fully supported for
injectable()andget_injected_obj()? - How does caching work?
- Is it production-ready?
A: If your project heavily relies on FastAPI's Depends() as the sole DI system and you don't want to introduce additional DI packages (like Dependency Injector or FastDepends), fastapi-injectable is your friend.
It allows you to reuse your existing FastAPI built-in DI system anywhere, without the need to refactor your entire codebase or maintain multiple DI systems.
Life is short, keep it simple!
A: You absolutely can if your situation allows you to:
- Modify large amounts of existing code that uses
Depends() - Maintain multiple DI systems in your project
fastapi-injectable focuses solely on extending FastAPI's built-in Depends() beyond routes. We're not trying to be another DI system - we're making the existing one more useful!
For projects with hundreds of dependency functions (especially with nested dependencies), this approach is more intuitive and requires minimal changes to your existing code.
Choose what works best for you!
A: Absolutely! That's exactly what this package was built for! fastapi-injectable was created to seamlessly work with FastAPI's dependency injection system, allowing you to reuse your existing Depends() code anywhere - not just in routes.
Focus on what matters instead of worrying about how to get your existing dependencies outside of FastAPI routes!
A: Yes! It supports:
- Regular dependencies
- Generator dependencies (with cleanup utility functions)
- Async dependencies
- Sync dependencies
- Nested dependencies (dependencies with sub-dependencies)
A: You have three options:
- Manual cleanup per function:
await cleanup_exit_stack_of_func(your_func) - Cleanup everything:
await cleanup_all_exit_stacks() - Automatic cleanup on shutdown:
setup_graceful_shutdown()
A: Yes! You can freely mix them. For running async code in sync contexts, use the provided run_coroutine_sync() utility.
A: Currently, type hint support is a work in progress. However, this doesn't affect the core benefits of the package (seamlessly reusing and maintaining consistency in your FastAPI DI system).
We're actively working on improving type hint support, and we'll have good news on this front soon! In the meantime, enjoy the elegant and clean solution that fastapi-injectable provides.
A: By default, dependencies are cached like in FastAPI routes. You can disable caching with @injectable(use_cache=False) if you need fresh instances.
A: Yes! The package has:
- 100% test coverage
- Type checking with
mypy - Comprehensive error handling
- Production use cases documented
Contributions are very welcome. To learn more, see the Contributor Guide.
Distributed under the terms of the MIT license,
fastapi-injectable is free and open source software.
If you encounter any problems, please file an issue along with a detailed description.
- This project was generated from @cjolowicz's Hypermodern Python Cookiecutter template.
- Thanks to @barapa's initiation, his work inspires me to create this project.
- [Issue] Using Depends() in other functions, outside the endpoint path operation!
- [Discussion] Using Depends() in other functions, outside the endpoint path operation!
My blog posts about the prototype of this project: