Skip to content

Dependencies

Propan uses the secondary library FastDepends for dependency management. This dependency system is literally borrowed from FastAPI, so if you know how to work with this framework, you know how to work with dependencies in Propan.

You can go to the documentation FastDepends if you want to get more details, however, the key points and additions will be covered here.

Type casting

The key function in the dependency management and type conversion system in Propan is the decorator @apply_types (@inject in FastDepends).

By default, it applies to all event handlers, unless you disabled the same option at a broker creation.

from propan import RedisBroker
broker = RedisBroker(..., apply_types=False)

from propan import RabbitBroker
broker = RabbitBroker(..., apply_types=False)

from propan import KafkaBroker
broker = KafkaBroker(..., apply_types=False)

from propan import SQSBroker
broker = SQSBroker(..., apply_types=False)

from propan import NatsBroker
broker = NatsBroker(..., apply_types=False)

Warning

By setting the apply_types=False flag, you disable not only type casting, but also Depends and Context.

This flag can be useful if you are using Propan within another framework and you do not need to use a native dependency system.

Dependency Injection

To implement dependencies in Propan, a special class Depends is used

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import PropanApp, RedisBroker, Depends

broker = RedisBroker("redis://localhost:6379")
app = PropanApp(broker)

def simple_dependency():
    return 1

@broker.handle("test")
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import PropanApp, RabbitBroker, Depends

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)

def simple_dependency():
    return 1

@broker.handle("test")
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import PropanApp, KafkaBroker, Depends

broker = KafkaBroker("localhost:9092")
app = PropanApp(broker)

def simple_dependency():
    return 1

@broker.handle("test")
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import PropanApp, SQSBroker, Depends

broker = SQSBroker("http://localhost:9324", ...)
app = PropanApp(broker)

def simple_dependency():
    return 1

@broker.handle("test")
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import PropanApp, NatsBroker, Depends

broker = NatsBroker("nats://localhost:4222")
app = PropanApp(broker)

def simple_dependency():
    return 1

@broker.handle("test")
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

The first step: we need to declare a dependency - it can be any Callable object.

Callable

"Callable" is an object that can be "called". It can be a function, a class, or a class method.

In other words: if you can write such code my_object() - my_object will be Callable

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

Second step: Declare which dependencies you need using Depends

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

10
11
async def handler(body: dict, d: int = Depends(simple_dependency)):
    assert d == 1

The last step: Just use the result of executing your dependency!

It's easy, isn't it?

Auto @apply_types

In the code above, we didn't use this decorator for our dependencies. However, it still applies to all functions used as dependencies. Keep this in your mind.

Top-level dependencies

If you don't need a dependency result you can use the following code:

@broker.handle("test")
def method(_ = Depends(...)): ...

But, using a special handle parameter is more suitable:

@broker.handle("test", dependencies=[Depends(...)])
def method(): ...

Also, you are able to declare broker-level dependencies: they will be applied to all brokers' handlers.

broker = RabbitBroker(dependencies=[Depends(...)])

Nested dependencies

Dependencies can also contain other dependencies. This works in a very predictable way: just declare Depends in the dependent function.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from propan import PropanApp, RedisBroker, Depends

broker = RedisBroker("redis://localhost:6379")
app = PropanApp(broker)

def another_dependency():
    return 1

def simple_dependency(b: int = Depends(another_dependency)): # (1)
    return b * 2

@broker.handle("test")
async def handler(
    body: dict,
    a: int = Depends(another_dependency),
    b: int = Depends(simple_dependency)):
    assert (a + b) == 3
  1. A nested dependency is called here
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from propan import PropanApp, RabbitBroker, Depends

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)

def another_dependency():
    return 1

def simple_dependency(b: int = Depends(another_dependency)): # (1)
    return b * 2

@broker.handle("test")
async def handler(
    body: dict,
    a: int = Depends(another_dependency),
    b: int = Depends(simple_dependency)):
    assert (a + b) == 3
  1. A nested dependency is called here
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from propan import PropanApp, KafkaBroker, Depends

broker = KafkaBroker("localhost:9092")
app = PropanApp(broker)

def another_dependency():
    return 1

def simple_dependency(b: int = Depends(another_dependency)): # (1)
    return b * 2

@broker.handle("test")
async def handler(
    body: dict,
    a: int = Depends(another_dependency),
    b: int = Depends(simple_dependency)):
    assert (a + b) == 3
  1. A nested dependency is called here
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from propan import PropanApp, SQSBroker, Depends

broker = SQSBroker("http://localhost:9324", ...)
app = PropanApp(broker)

def another_dependency():
    return 1

def simple_dependency(b: int = Depends(another_dependency)): # (1)
    return b * 2

@broker.handle("test")
async def handler(
    body: dict,
    a: int = Depends(another_dependency),
    b: int = Depends(simple_dependency)):
    assert (a + b) == 3
  1. A nested dependency is called here
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from propan import PropanApp, NatsBroker, Depends

broker = NatsBroker("nats://localhost:4222")
app = PropanApp(broker)

def another_dependency():
    return 1

def simple_dependency(b: int = Depends(another_dependency)): # (1)
    return b * 2

@broker.handle("test")
async def handler(
    body: dict,
    a: int = Depends(another_dependency),
    b: int = Depends(simple_dependency)):
    assert (a + b) == 3
  1. A nested dependency is called here

Caching

In the example above, the another_dependency function will be called at ONCE!. Propan caches all dependency execution results within ONE @apply_stack call stack. This means that all nested dependencies will receive the cached result of dependency execution. But, between different calls of the main function, these results will be different.

To prevent this behavior, just use Depends(..., cache=False). In this case, the dependency will be used for each function in the call stack where it is used.

Use with regular functions

You can use the decorator @apply_types not only together with your `@broker.handle', but also with the usual functions: both synchronous and asynchronous.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from propan import Depends, apply_types

def simple_dependency(a: int, b: int = 3):
    return a + b

@apply_types
def method(a: int, d: int = Depends(simple_dependency)):
    return a + d

assert method("1") == 5

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import asyncio
from propan import Depends, apply_types

async def simple_dependency(a: int, b: int = 3):
    return a + b

def another_dependency(a: int):
    return a

@apply_types
async def method(
    a: int,
    b: int = Depends(simple_dependency),
    c: int = Depends(another_dependency),
):
    return a + b + c

assert asyncio.run(method("1")) == 6

Be careful

In asynchronous code, you can use both synchronous and asynchronous dependencies. But in synchronous code, only synchronous dependencies are available to you.

Casting dependency types

FastDepends, used by Propan, also gives the type return. This means that the value returned by the dependency will be be cast to the type twice: as return these are dependencies and as the input argument of the main function. This does not incur additional costs if these types have the same annotation. Just keep it in mind. Or not... Anyway, I've warned you.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from propan import Depends, apply_types

def simple_dependency(a: int, b: int = 3) -> str:
    return a + b  # 'return' is cast to `str` for the first time

@inject
def method(a: int, d: int = Depends(simple_dependency)):
    # 'd' is cast to `int` for the second time
    return a + d

assert method("1") == 5

Also, the result of executing the dependency is cached. If you use this dependency in N functions, this cached result will be converted to type N times (at the input to the function being used).

To avoid problems with this, use mypy or just be careful with the annotation of types in your project.