Easily Reusing Depends Outside FastAPI Routes

Reuse Your Dependencies in Workers, CLI Tools, Development Tools, and Testing

October 26, 2024

🚀 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!

  1. CLI Tools
  2. Background Workers
  3. REPL Server
  4. 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:

  1. Python: 3.12.4
  2. 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:

  1. Worker
  2. Command-line Tool (CLI)
  3. Testing
  4. 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.

Tiangolo

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:

  1. Implements a decorator
  2. Uses fastapi.dependencies.utils.get_dependant to transform the decorated func (async) function into a Dependant instance, parsing dependencies.
  3. 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:

  1. Doesn’t work on synchronous functions
  2. Lacks support for use_cache in Depends(), 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 in Depends(), 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