Зависимости¶

Propan использует второстепенную библиотеку FastDepends ддя управления зависимостями. Эта система зависимостей буквально позаимствована у FastAPI, так что, если вы умеет работать с этим фреймворков - вы умеете работать с зависимостями в Propan.

Вы можете перейти в документацию FastDepends, если хотите получить больше подробностей, однако, ключевые моменты и дополнения будут освещены здесь.

Приведение типов¶

Ключевой функцией в системе управления зависимостями и приведения типов в Propan является декоратор @apply_types (@inject в FastDepends).

По умолчанию он применяется ко всем обработчикам событий, если только вы не отключили соответсвующую опцию при создании брокера.

RedisRabbitMQKafkaSQSNATS

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

Выставив флаг apply_types=False вы отключаете не только приведение типов, но и Depends, Context.

Этот флаг может быть полезен, если вы используете Propan в рамках другого фреймворка и вам не нужно использовать нативную систему зависимостей.

Внедрение зависимостей¶

Для внедрения зависимостей в Propan используется специальный класс Depends

RedisRabbitMQKafkaSQSNATS

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

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

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

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

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

Первым шагом: нам нужно объявить зависимость - это может быть любой Callable объект.

Callable

"Callable" - объект, который может быть "вызван". Это может быть функция, класс или метод класса.

Другими словами: если вы можете написать такой код my_object() - my_object будет Callable

RedisRabbitMQKafkaSQSNATS

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

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

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

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

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

Вторым шагом: объявите, какие зависимости вам нужны с помощью Depends

RedisRabbitMQKafkaSQSNATS

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

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

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

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

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

Последним шагом: просто используйте результат выполнения вашей зависимости!

Это ведь просто, разве нет?

Автоматическое применений @apply_types

В коде выше мы не использовали этот декоратор для наших зависимостей. Однако, он все равно применяется ко всем функциям, используемым в качестве зависимостей.

Зависимости верхнего уровня¶

Если вам не нужен результат выполнения зависимостей, вы, конечно, можете использовать следующую конструкцию:

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

Однако, гораздо удобнее использовать для этого специальный параметр метода handle

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

Также вы можете объявить такие зависимости на уровне брокера: в таком случае, они будут применяться ко всем обработчикам этого брокера.

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

Вложенные зависимости¶

Зависимости также могут содержать другие зависимости. Это работает очень предсказуемым образом: просто объявите Depends в зависимой функции.

RedisRabbitMQKafkaSQSNATS

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

Здесь вызывается вложенная зависимость

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

Здесь вызывается вложенная зависимость

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

Здесь вызывается вложенная зависимость

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

Здесь вызывается вложенная зависимость

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

Здесь вызывается вложенная зависимость

Кеширование

В примере выше функция another_dependency будет вызвана ОДИН РАЗ!. Propan кеширует все результаты выполнения зависимостей в рамках ОДНОГО @apply_stack стека вызова. Это означает, что все вложенные зависимости получат закешированный результат выполнения зависимости. Но, между разными вызовами основной функции, эти результаты будут различными.

Чтобы предотвратить это поведение, просто используйте Depends(..., cache=False). В этом случае зависимость будет испольняться для каждой функции в стеке вызова, где она используется.

Использование с обычными функциями¶

Вы можете использовать декоратор @apply_types не только вместе с вашими @broker.handle'ми, но и с обычными функциями: как синхронными, так и асинхронными.

SyncAsync

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

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

Будьте аккуратны

В асинхронном коде вы можете использовать как синхронные, так и асинхронные зависимости. Но в синхронном коде вам доступны только синхронные зависимости.

Приведение типов зависимостей¶

FastDepends, используемый Propan, также приводит тип return. Это означает, что значение, возвращаемое зависимостью будет дважды приводиться к типу: как return этой зависимости и как входной аргумент основной функции. Это не несет дополнительных расходов, если эти типы имеют одну и ту же аннотацию. Просто держите это в голове. Или нет... В любом случае, я вас предупредил.

from propan import Depends, apply_types

def simple_dependency(a: int, b: int = 3) -> str:
    return a + b  # 'return' приводится к `str` в первый раз

@inject
def method(a: int, d: int = Depends(simple_dependency)):
    # 'd' приводится к `int` во второй раз
    return a + d

assert method("1") == 5

Также, результат выполнения зависимости кешируется. Если вы используете эту зависимости в N функциях, этот закешированный результат будет приводится к типу N раз (на входе в используемую функцию).

Чтобы избежать потенциальных проблем, используйте mypy или просто будьте аккуратны с аннотацией типов в вашем проекте.