CONTEXT¶

Propan stores the context of the application and each request. You can access them using a special class Context.

from propan import Context

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

Existing fields¶

Context already contains some global objects that you can always access:

app - the PropanApp object of your application
broker - current broker
context - the context itself, in which you can write your own fields
logger - logger used for your broker (tags messages with message_id)
message - raw message (if you need access to it)

At the same time, thanks to contextlib.ContextVar, message always corresponds to the context the current handler process.

Access to context fields¶

By default, as in the example above, the context searches for an object based on the argument name.

from propan import Context

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

Access by name¶

Sometimes you may need to use a different name for the argument (not the one under which it is stored in the context). Or even get access not to the whole object, but only to its field or method. To do this, just specify by name what you want to get - and the context will provide you with the wished object.

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¶

The default method is not very convenient if you need to use the same context field throughout the project. Also, it requires explicit annotation of the type of the incoming argument if we want to use the auto-completion of our IDE. In order to avoid long import chains and code duplication, Context is fully compatible with 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,
):
    ...

For your convenience, Propan already contains annotations for existing context fields. You can import them and use at your code.

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,
):
    ...

Default values¶

If you try to access a field that does not exist in the global context, you will get the pydantic.ValidationError exception.

However, you can set the default value if you feel the need.

from propan import Context

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

Casting context types¶

By default, context fields are NOT CAST to the type specified in their annotation. If you need this functionality, you can set the appropriate flag.

from propan import Context

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

Declaration of context fields¶

Globally¶

To declare the context fields, you need to call the context.set_global method with an indication of the key by which the object will be placed in the context.

from propan.annotations import ContextRepo

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

In this case, the field becomes a global context field: it does not depend on the current message handler (unlike message)

To remove a field from the context use reset_global

context.reset_global("my_key")

Locally¶

To set the local context (it will act in all functions called inside it), use the context manager 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

Also, you can set the context yourself: then it will act within the current call stack until you clear it.

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

Use in other functions¶

By default, the context is available in the same place as Depends:

at life cycle hooks
message handlers
dependencies

Depends¶

When using Context in Depends, there is no need to write additional code: like nested Depends, Context is also available by default.

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

Normal functions¶

To use context at other functions use the decorator @apply_types. This case, the called function context will correspond to the context of the event handler from which it was called.

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)

In the example above, we did not pass the logger function at calling, it was placed out of context.