Skip to content

Application Context#

FastStreams has its own Dependency Injection container - Context, used to store application runtime objects and variables.

With this container, you can access both application scope and message processing scope objects. This functionality is similar to Depends usage.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from faststream import Context, FastStream
from faststream.kafka import KafkaBroker

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


@broker.subscriber("test")
async def base_handler(
    body: str,
    message=Context(),  # get access to raw message
):
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from faststream import Context, FastStream
from faststream.rabbit import RabbitBroker

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


@broker.subscriber("test")
async def base_handler(
    body: str,
    message=Context(),  # get access to raw message
):
    ...

But, with the Annotated Python feature usage, it is much closer to @pytest.fixture.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from typing import Annotated

from faststream import Context, FastStream
from faststream.kafka import KafkaBroker, KafkaMessage

Message = Annotated[KafkaMessage, Context()]

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


@broker.subscriber("test")
async def base_handler(
    body: str,
    message: Message,  # get access to raw message
):
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from typing import Annotated

from faststream import Context, FastStream
from faststream.rabbit import RabbitBroker, RabbitMessage

Message = Annotated[RabbitMessage, Context()]

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


@broker.subscriber("test")
async def base_handler(
    body: str,
    message: Message,  # get access to raw message
):
    ...

Usages#

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

  • at lifespan hooks
  • message subscribers
  • nested dependencies

Tip

Fields obtained from the Context are editable, so editing them in a function means editing them everywhere.

Compatibility with Regular Functions#

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import Context, apply_types


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


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

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

Last update: 2023-09-24