CONTEXT¶

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

from propan import Context

@broker.hanlde("test")
async def handler(broker = Context()):
    await broker.publish("response", "response-queue")

Существующие поля¶

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

app - объект PropanApp вашего приложения
broker - текущий брокер
context - непосредственно сам контекс, в который вы можете записать собственные поля
logger - logger, используемый для вашего брокера (помечает сообщения с помощью message_id)
message - необработанное сообщение (если вам нужен доступ к нему)

При этом, благодаря contextlib.ContextVar, message всегда соответствует контексту текущего процесса обработки.

Доступ к полям контекста¶

По умолчанию, как в примере выше, контекст ищет объект исходя из названия аргумента.

from propan import Context

@broker.hanlde("test")
async def handler(
    body: dict,
    app = Context(),
    broker = Context(),
    context = Context(),
    logger = Context(),
    message = Context(),
):
    ...

Доступ по имени¶

Иногда у вас может возникнуть необходимость использовать другое название для аргумента (не то, под которым он хранится в контексте). Или даже получить доступ не ко всему объекту, а только к его полю или методу. Для этого просто укажите по имени, что вы хотите получить - и контекст предоставит вам нужный объект.

from propan import Context

@broker.hanlde("test")
async def handler(
    body: dict,
    propan_app = Context("app"),
    publish = Context("broker.publish"),
    secret_key = Context("settings.app.secret_key"),
):
    await publish(secret_key, "secret-queue")

Annotated¶

Способ по умолчанию не слишком удобен, если вам необходимо использовать одно и то же поле контекста по всему проекту. Также, оно требует явного указания аннотации типа входящего аргумента, если мы хотим пользоваться автодополнением нашей IDE. Для того, чтобы избежать длинных цепочек импортов и дублирования кода, Context полностью совместим с typing.Annotated.

from propan import Context, RabbitBroker
from typing_extension import Annotated

Broker = Annotated[RabbitBroker, Context("broker")]

@broker.hanlde("test")
async def handler(
    body: dict,
    broker: Broker,
):
    ...

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

from propan import annotations

@rabbit_broker.handle("test")
async def base_handler(
    body: dict,
    app: annotations.App,
    context: annotations.ContextRepo,
    logger: annotations.Logger,
    broker: annotations.RabbitBroker,
    message: annotations.RabbitMessage,
):
    ...

Значения по умолчанию¶

Если же вы попробуете получить доступ к полю, которого нет в глобальном контексте, вы получите ошибку pydantic.ValidationError.

Однако, вы можете установить значение по умолчанию, если испытываете в этом необходимость.

from propan import Context

@broker.hanlde("test")
async def handler(
    body: dict,
    some_field = Context(default=None)
):
    assert some_field is None

Приведение типов контекста¶

По умолчанию, поля контекста НЕ ПРИВОДЯТСЯ к типу, указанному в их аннотации. Если вам необходим этот функционал, вы просто можете установить соотвествующий флаг.

from propan import Context

@broker.hanlde("test")
async def handler(
    body: dict,
    some_field: int = Context(default="1", cast=True)
):
    assert some_field == 1

Объявление полей контекста¶

Глобально¶

Для объявления полей контекста необходимо просто вызвать метод context.set_global с указанием ключа, по которому объект будет помещен в контекст.

from propan.annotations import ContextRepo

@broker.hanlde("test")
async def handler(
    body: dict,
    context: ContextRepo
):
    context.set_global("my_key", 1)

При этом поле становится глобальным полем контекста: оно не зависит от текущего обработчика сообщения (в отличие от message)

Для удаления поля из контекста просто используйте reset_global

context.reset_global("my_key")

Локально¶

Для установки локального контекста (он будет действовать во всех функциях, вызванных внутри него) используйте контекстный менеджер scope

from propan import apply_types, Context
from propan.annotations import ContextRepo

@broker.hanlde("test")
async def handler(
    body: dict,
    context: ContextRepo
):
    with context.scope("local", 1):
        nested_function()

@apply_types
def nested_function(local = Context()):
    assert local == 1

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

from propan import apply_types, Context
from propan.annotations import ContextRepo

@broker.hanlde("test")
async def handler(
    body: dict,
    context: ContextRepo
):
    token = context.set_local("local", 1):
    nested_function()
    context.reset_local("local", token)

@apply_types
def nested_function(local = Context()):
    assert local == 1

Использование в других функциях¶

По умолчанию контекст доступен там же, где и Depends:

в хуках жизненного цикла
обработчиках сообщений
зависимостях

Depends¶

При использовании Context в Depends нет необходимости писать дополнительный код: как и вложенные Depends, Context также доступен по умолчанию.

from propan import Context, Depends

def nested_func(
    body: dict,
    logger = Context()
):
    logger.info(body)
    return body

@broker.hanlde("test")
async def handler(body: dict, n = Depends(nested_func)):
    pass

Обычные функции¶

Если же вы хотите использовать контекст и в других функциях, просто используйте декоратор @apply_types. При этом контекст вызванной функции будет соответсвовать контексту обработчика события, из которого она вызвана.

from propan import apply_types, Context

@broker.hanlde("test")
async def handler(body: dict):
    nested_func()

@apply_types
def nested_func(
    body: dict,
    logger = Context()
):
    logger.info(body)

В примере выше мы не передавали при вызове функции logger, он был помещен из контекста.