🚀 2024/12/21 Important Update 🚀
It’s been two months since the original post, and I’d like to share my first open-source package - fastapi-injectable.
The reason for creating fastapi-injectable
is to address the pain points mentioned in the original post. It’s a production-ready package that allows you to seamlessly use code with Depends()
parameters outside of FastAPI routes. Let’s skip the talk and dive straight into the code:
Before using it, install it via pip install fastapi-injectable
!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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'
Before fastapi-injectable
, it was not easy to quickly reuse existing code. If your entire project is based on FastAPI Depends()
for business logic, you had to create many workarounds because Depends()
couldn’t be used outside FastAPI routes.
But now, that’s no longer an issue! With fastapi-injectable
, all those tricky scenarios have been resolved!
- CLI Tools
- Background Workers
- REPL Server
- Anywhere outside FastAPI Routes
If you have any feedback or ideas, feel free to open an issue or submit a PR to help improve fastapi-injectable. I’d also appreciate it if you could star the project ✨. It would make me really happy!
Continuing from the original post…
Versions
The code in this article has been tested in the following environment:
- Python:
3.12.4
- FastAPI:
0.115.0
Background
In FastAPI projects, we often use dependencies with Depends(get_your_dependant)
to decouple code through dependency injection. However, outside the FastAPI app, the standard Depends
function cannot be used directly with pure Python alone. This means if you want to reuse code like this in:
- Worker
- Command-line Tool (CLI)
- Testing
- Development Tool
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import Annotated
from fastapi import Depends
class Brand:
pass
class Car:
def __init__(self, brand: Brand) -> None:
self.brand = brand
def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
brand = get_brand() # No Error since this function doesn't have any dependencies
car = get_car() # TypeError: get_car() missing 1 required positional argument: 'brand'
…it won’t work.
Annotated
is a type annotation; more on that can be found in the Python Docs. By itself, Annotated
does nothing in Python—it’s there for the user to handle. In our case, FastAPI acts as the “user” of this annotation, processing dependencies when it finds an Annotated
field with Depends
. Without FastAPI, the following get_car()
function in pure Python is equivalent to:
1
2
3
4
5
# Oops, no one is handling the `brand` for us
def get_car(brand: Brand) -> Car:
return Car(brand)
car = get_car() # No wonder we will get TypeError: get_car() missing 1 required positional argument: 'brand'
The Issue
Does this mean all the Depends()
in a project can’t be reused directly? Does it mean every single get_your_dependant()
has to be rewritten? That would be inconvenient!
I’m not the only one who’s run into this issue; it’s been discussed in a FastAPI Github thread – Using Depends() in other functions, outside the endpoint path operation! #7720.
FastAPI’s creator, Tiangolo, suggests that to use these objects outside of FastAPI, you should avoid the extra complexity of FastAPI and instead explicitly instantiate objects and pass them in. However, this approach has not been widely accepted by the community.
I understand his viewpoint; after all, “Explicit is better than implicit” is part of the Zen of Python. But in real-world applications, the benefit of reusing a lot of existing code quickly outweighs the cost of strictly following this convention (Z>B).
Other Solutions
Most solutions suggest ways to reuse Depends
code in Testing, Workers, CLI, etc., often recommending alternative DI frameworks like dependency-injector, injector, and pinject. However, these options fall short since they’d require significant rewriting, failing to meet the core need: reusing Depends
without hassle.
Then I found @barapa’s solution, which I’ll expand on below. I’m grateful for their meaningful contribution!
@barapa’s Version
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
import inspect
from contextlib import AsyncExitStack
from functools import wraps
from typing import Any, Callable, Coroutine, TypeVar
from fastapi import Request
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import get_dependant, solve_dependencies
from loguru import logger
T = TypeVar("T")
class DependencyError(Exception):
"""Exception raised for errors during dependency injection."""
pass
def injectable(
func: Callable[..., Coroutine[Any, Any, T]],
) -> Callable[..., Coroutine[Any, Any, T]]:
"""
A decorator to enable FastAPI-style dependency injection for any asynchronous function.
This allows dependencies defined with FastAPI's Depends mechanism to be automatically
resolved and injected into CLI tools or other components, not just web endpoints.
Args:
func: The asynchronous function to be wrapped, enabling dependency injection.
Returns:
The wrapped function with dependencies injected.
Raises:
ValueError: If the dependant.call is not a callable function.
RuntimeError: If the wrapped function is not asynchronous.
DependencyError: If an error occurs during dependency resolution.
"""
@wraps(func)
async def call_with_solved_dependencies(*args: Any, **kwargs: Any) -> T: # type: ignore
dependant: Dependant = get_dependant(path="command", call=func)
if dependant.call is None or not callable(dependant.call):
raise ValueError("The dependant.call attribute must be a callable.")
if not inspect.iscoroutinefunction(dependant.call):
raise RuntimeError("The decorated function must be asynchronous.")
fake_request = Request({"type": "http", "headers": [], "query_string": ""})
values: dict[str, Any] = {}
errors: list[Any] = []
async with AsyncExitStack() as stack:
solved_result = await solve_dependencies(
request=fake_request,
dependant=dependant,
async_exit_stack=stack,
)
values, errors = solved_result[0], solved_result[1]
if errors:
error_details = "\n".join([str(error) for error in errors])
logger.info(f"Dependency resolution errors: {error_details}")
return await dependant.call(*args, **{**values, **kwargs})
return call_with_solved_dependencies
from cyclopts import App
from fastapi import Depends
from loguru import logger
from app.clis.lib.injectable import injectable
from app.deps.settings import Settings, provide_settings
app = App()
@app.default
@injectable
async def example(
*,
message: str,
settings: Settings = Depends(provide_settings),
) -> None:
"""Example command using injectable with cyclopts"""
logger.info(message)
logger.warning(settings.generate_db_url())
if __name__ == "__main__":
app()
In short, this solution:
- Implements a decorator
- Uses
fastapi.dependencies.utils.get_dependant
to transform the decoratedfunc
(async) function into aDependant
instance, parsing dependencies. - Upon success, directly calls the original
func
and passes the parsed dependencies as arguments.
This solution cleverly leverages FastAPI’s public utility functions to accomplish the goal without needing private methods or variables, minimizing the risk of breaking changes in future updates.
However, I wanted to make a few improvements to increase usability for my project. Currently, it has some limitations:
- Doesn’t work on synchronous functions
- Lacks support for
use_cache
inDepends()
, which controls Singleton use
My Solution
To meet my project needs, I modified it to address both limitations. Although supporting both sync and async functions in a decorator goes against SRP and Python convention, in this scenario—working outside of a FastAPI web app in tools like Workers and CLI — it’s worth the tradeoff.
Doesn’t work on synchronous functions
The fix was simple: have the decorator check if the input function is synchronous or asynchronous using inspect.iscoroutinefunction
.
Lacks support for
use_cache
inDepends()
, which controls Singleton use
I added a use_cache
parameter to the decorator maker function, following solve_dependencies
’ handling of dependency_cache
.
When use_cache=True
A global dictionary _SOLVED_DEPENDENCIES
holds parsed dependencies. If a dependency is already cached, solve_dependencies
retrieves it without reinitializing.
When use_cache=False
A new dependency_cache
dictionary is created each time to ensure dependencies are always re-solved.
Complete Code
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
import asyncio
import inspect
import logging
from collections.abc import Callable, Coroutine
from contextlib import AsyncExitStack
from functools import wraps
from typing import Any, TypeVar, cast
from fastapi import Request
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import get_dependant, solve_dependencies
logger = logging.getLogger(__name__)
T = TypeVar("T")
_SOLVED_DEPENDENCIES: dict[tuple[Callable[..., Any], tuple[str]], Any] = {}
class DependencyError(Exception):
"""Exception raised for errors during dependency injection."""
def injectable( # noqa: C901
func: Callable[..., T] | Callable[..., Coroutine[Any, Any, T]] | None = None,
*,
use_cache: bool = True,
) -> Callable[..., T] | Callable[..., Coroutine[Any, Any, T]]:
"""A decorator to enable FastAPI-style dependency injection for any function (sync or async).
This allows dependencies defined with FastAPI's Depends mechanism to be automatically
resolved and injected into CLI tools or other components, not just web endpoints.
Args:
func: The function to be wrapped, enabling dependency injection.
use_cache: Whether to use the dependency cache for the arguments a.k.a sub-dependencies.
Returns:
The wrapped function with dependencies injected.
Raises:
ValueError: If the dependant.call is not a callable function.
DependencyError: If an error occurs during dependency resolution.
"""
def _impl(
func: Callable[..., T] | Callable[..., Coroutine[Any, Any, T]],
) -> Callable[..., T] | Callable[..., Coroutine[Any, Any, T]]:
is_async = inspect.iscoroutinefunction(func)
dependency_cache = _SOLVED_DEPENDENCIES if use_cache else None
async def resolve_dependencies(dependant: Dependant) -> tuple[dict[str, Any], list[Any] | None]:
fake_request = Request({"type": "http", "headers": [], "query_string": ""})
async with AsyncExitStack() as stack:
solved_result = await solve_dependencies(
request=fake_request,
dependant=dependant,
async_exit_stack=stack,
embed_body_fields=False,
dependency_cache=dependency_cache,
)
dep_kwargs = solved_result.values # noqa: PD011
if dependency_cache is not None:
dependency_cache.update(solved_result.dependency_cache)
return dep_kwargs, solved_result.errors
def handle_errors(errors: list[Any] | None) -> None:
if errors:
error_details = "\n".join(str(error) for error in errors)
logger.info(f"Dependency resolution errors: {error_details}")
def validate_dependant(dependant: Dependant) -> None:
if dependant.call is None or not callable(dependant.call):
msg = "The dependant.call attribute must be a callable."
raise ValueError(msg)
@wraps(func)
async def async_call_with_solved_dependencies(*args: Any, **kwargs: Any) -> T: # noqa: ANN401
dependant = get_dependant(path="command", call=func)
validate_dependant(dependant)
deps, errors = await resolve_dependencies(dependant)
handle_errors(errors)
return await cast(Callable[..., Coroutine[Any, Any, T]], dependant.call)(*args, **{**deps, **kwargs})
@wraps(func)
def sync_call_with_solved_dependencies(*args: Any, **kwargs: Any) -> T: # noqa: ANN401
dependant = get_dependant(path="command", call=func)
validate_dependant(dependant)
deps, errors = asyncio.run(resolve_dependencies(dependant))
handle_errors(errors)
return cast(Callable[..., T], dependant.call)(*args, **{**deps, **kwargs})
return async_call_with_solved_dependencies if is_async else sync_call_with_solved_dependencies
if func is None:
return _impl # type: ignore # noqa: PGH003
return _impl(func)
Usage Example
Let’s use the previous Car
and Brand
example:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import Annotated
from fastapi import Depends
class Brand:
pass
class Car:
def __init__(self, brand: Brand) -> None:
self.brand = brand
def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
brand = get_brand() # No Error since this function doesn't have any dependencies
car = get_car() # TypeError: get_car() missing 1 required positional argument: 'brand'
We now have two ways to achieve our goal: you can use @injectable
to decorate a function with Depends
parameters, or wrap it directly with injectable(func)
.
1. Using @injectable
as a Decorator
use_cache=True
(Default)
1
2
3
4
5
6
7
8
@injectable # Equals to @injectable(use_cache=True)
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car() # <__main__.Car at 0x10620cf20>
car_2 = get_car() # <__main__.Car at 0x10953c140>
car_3 = get_car() # <__main__.Car at 0x1095b7e00>
assert car_1.brand is car_2.brand is car_3.brand # True
Note: Each Car
instance shares the same Brand
instance.
use_cache=False
1
2
3
4
5
6
7
8
@injectable(use_cache=False)
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car() # <__main__.Car at 0x10620cf20>
car_2 = get_car() # <__main__.Car at 0x10953c140>
car_3 = get_car() # <__main__.Car at 0x1095b7e00>
assert car_1.brand is not car_2.brand is not car_3.brand # True
Each Car
instance has a unique Brand
instance.
2. Using injectable(func)
as a Wrapper
use_cache=True
(Default)
1
2
3
4
5
6
7
8
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car) # Equals to injectable(get_car, use_cache=True)
car_1 = injectable_get_car() # <__main__.Car at 0x10620cf20>
car_2 = injectable_get_car() # <__main__.Car at 0x10953c140>
car_3 = injectable_get_car() # <__main__.Car at 0x1095b7e00>
assert car_1.brand is car_2.brand is car_3.brand # True
use_cache=False
1
2
3
4
5
6
7
8
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car, use_cache=False)
car_1 = injectable_get_car() # <__main__.Car at 0x10620cf20>
car_2 = injectable_get_car() # <__main__.Car at 0x10953c140>
car_3 = injectable_get_car() # <__main__.Car at 0x1095b7e00>
assert car_1.brand is not car_2.brand is not car_3.brand # True
Plus: Complete Test Case
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
# type: ignore # noqa: PGH003
from typing import Annotated
from unittest.mock import Mock, patch
import pytest
from fastapi import Depends
from fastapi.dependencies.models import Dependant
# (Paste the injectable here)
# def injectable(...):
# ...
class Brand:
pass
class Car:
def __init__(self, brand: Brand) -> None:
self.brand = brand
def test_injectable_sync_only_decorator_with_cache() -> None:
def get_brand() -> Brand:
return Brand()
@injectable
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car()
car_2 = get_car()
car_3 = get_car()
assert car_1.brand is car_2.brand is car_3.brand
def test_injectable_sync_only_decorator_with_no_cache() -> None:
def get_brand() -> Brand:
return Brand()
@injectable(use_cache=False)
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car()
car_2 = get_car()
car_3 = get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
def test_injectable_sync_only_wrap_function_with_cache() -> None:
def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car)
car_1 = injectable_get_car()
car_2 = injectable_get_car()
car_3 = injectable_get_car()
assert car_1.brand is car_2.brand is car_3.brand
def test_injectable_sync_only_wrap_function_with_no_cache() -> None:
def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car, use_cache=False)
car_1 = injectable_get_car()
car_2 = injectable_get_car()
car_3 = injectable_get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
async def test_injectable_async_only_decorator_with_cache() -> None:
async def get_brand() -> Brand:
return Brand()
@injectable
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = await get_car()
car_2 = await get_car()
car_3 = await get_car()
assert car_1.brand is car_2.brand is car_3.brand
async def test_injectable_async_only_decorator_with_no_cache() -> None:
async def get_brand() -> Brand:
return Brand()
@injectable(use_cache=False)
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = await get_car()
car_2 = await get_car()
car_3 = await get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
async def test_injectable_async_only_wrap_function_with_cache() -> None:
async def get_brand() -> Brand:
return Brand()
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car)
car_1 = await injectable_get_car()
car_2 = await injectable_get_car()
car_3 = await injectable_get_car()
assert car_1.brand is car_2.brand is car_3.brand
async def test_injectable_async_only_wrap_function_with_no_cache() -> None:
async def get_brand() -> Brand:
return Brand()
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car, use_cache=False)
car_1 = await injectable_get_car()
car_2 = await injectable_get_car()
car_3 = await injectable_get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
async def test_injectable_async_with_sync_decorator_with_cache() -> None:
def get_brand() -> Brand:
return Brand()
@injectable
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = await get_car()
car_2 = await get_car()
car_3 = await get_car()
assert car_1.brand is car_2.brand is car_3.brand
async def test_injectable_async_with_sync_decorator_with_no_cache() -> None:
def get_brand() -> Brand:
return Brand()
@injectable(use_cache=False)
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = await get_car()
car_2 = await get_car()
car_3 = await get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
async def test_injectable_async_with_sync_wrap_function_with_cache() -> None:
def get_brand() -> Brand:
return Brand()
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car)
car_1 = await injectable_get_car()
car_2 = await injectable_get_car()
car_3 = await injectable_get_car()
assert car_1.brand is car_2.brand is car_3.brand
async def test_injectable_async_with_sync_wrap_function_with_no_cache() -> None:
def get_brand() -> Brand:
return Brand()
async def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car, use_cache=False)
car_1 = await injectable_get_car()
car_2 = await injectable_get_car()
car_3 = await injectable_get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
def test_injectable_sync_with_async_decorator_with_cache() -> None:
async def get_brand() -> Brand:
return Brand()
@injectable
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car()
car_2 = get_car()
car_3 = get_car()
assert car_1.brand is car_2.brand is car_3.brand
def test_injectable_sync_with_async_decorator_with_no_cache() -> None:
async def get_brand() -> Brand:
return Brand()
@injectable(use_cache=False)
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
car_1 = get_car()
car_2 = get_car()
car_3 = get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
def test_injectable_sync_with_async_wrap_function_with_cache() -> None:
async def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car)
car_1 = injectable_get_car()
car_2 = injectable_get_car()
car_3 = injectable_get_car()
assert car_1.brand is car_2.brand is car_3.brand
def test_injectable_sync_with_async_wrap_function_with_no_cache() -> None:
async def get_brand() -> Brand:
return Brand()
def get_car(brand: Annotated[Brand, Depends(get_brand)]) -> Car:
return Car(brand)
injectable_get_car = injectable(get_car, use_cache=False)
car_1 = injectable_get_car()
car_2 = injectable_get_car()
car_3 = injectable_get_car()
assert car_1.brand is not car_2.brand is not car_3.brand
Known Issues
This solution meets 95% of my needs, greatly improving my development experience. However, the injectable
function includes several type: ignore
and noqa
comments, suggesting type annotation issues that might not pass with static type checkers like mypy
or pyright
.
The reasons include the combined handling of Callable[..., T]
and Callable[..., Awaitable[T]]
, which is not fully supported by type checkers. For now, though, it fully meets my project’s requirements.
Summary
Having primarily worked with Django and DRF, I haven’t delved as deeply into FastAPI until now. Gaining insight into its design philosophy has been enlightening.
I’m glad I had the chance to tackle this real-world issue, and I hope this tool helps you too!
References
FastAPI Dependencies
Using Depends() in other functions, outside the endpoint path operation! #7720
PEP 492 – Coroutines with async and await syntax