Access to Message Information#

As you know, FastStream serializes a message body and provides you access to it through function arguments. But sometimes you need access to a message_id, headers, or other meta-information.

Message Access#

You can get it in a simple way: just access the message object in the Context.

This message contains the required information such as:

• body: bytes
• decoded_body: Any
• content_type: str
• reply_to: str
• headers: dict[str, Any]
• path: dict[str, Any]
• message_id: str
• correlation_id: str

Also, it is a FastStream wrapper around a native broker library message (aio_pika.IncomingMessage in the RabbitMQ case) that you can access with raw_message.

from faststream.rabbit import RabbitMessage

@broker.subscriber("test")
async def base_handler(
    body: str,
    msg: RabbitMessage,
):
    print(msg.correlation_id)

Also, if you can't find the information you require, you can get access directly to the wrapped aio_pika.IncomingMessage, which contains complete message information.

from aio_pika import IncomingMessage
from faststream.rabbit import RabbitMessage

@broker.subscriber("test")
async def base_handler(body: str, msg: RabbitMessage):
    raw: IncomingMessage = msg.raw_message
    print(raw)

Message Fields Access#

But in most cases, you don't need all message fields; you need to access some of them. You can use Context Fields access feature for this reason.

For example, you can access the correlation_id like this:

from faststream import Context

@broker.subscriber("test")
async def base_handler(
    body: str,
    cor_id: str = Context("message.correlation_id"),
):
    print(cor_id)

Or even directly from the raw message:

from faststream import Context

@broker.subscriber("test")
async def base_handler(
    body: str,
    cor_id: str = Context("message.raw_message.correlation_id"),
):
    print(cor_id)

But this code is too long to be reused everywhere. In this case, you can use a Python Annotated feature:

from types import Annotated
from faststream import Context

CorrelationId = Annotated[str, Context("message.correlation_id")]

@broker.subscriber("test")
async def base_handler(
    body: str,
    cor_id: CorrelationId,
):
    print(cor_id)

from typing_extensions import Annotated
from faststream import Context

CorrelationId = Annotated[str, Context("message.correlation_id")]

@broker.subscriber("test")
async def base_handler(
    body: str,
    cor_id: CorrelationId,
):
    print(cor_id)

Headers Access#

Sure, you can get access to a raw message and get the headers dict itself, but more often you just need a single header field. So, you can easily access it using the Context:

from faststream import Context

@broker.subscriber("test")
async def base_handler(
    body: str,
    user: str = Context("message.headers.user"),
):
    ...

Using the special Header class is more convenient, as it also validates the header value using Pydantic. It works the same way as Context, but it is just a shortcut to Context with a default setup. So, you already know how to use it:

from faststream import Header

@broker.subscriber("test")
async def base_handler(
    body: str,
    user: str = Header(),
):
    ...

Topic Pattern Access#

As you know, Rabbit allows you to use a pattern like this "logs.*" with a Topic exchange. Getting access to the real * value is an often-used scenario and FastStream provide it to you with the Path object (which is a shortcut to Context("message.path.*")).

To use it, you just need to replace your * with {variable-name} and use Path as a regular Context object:

from faststream import Path
from faststream import RabbitQueue, RabbitExchane, ExchangeType

@broker.subscriber(
    RabbitQueue(
        "test-queue",
        routing_key="logs.{level}",
    ),
    RabbitExchange(
        "test-exchange",
        type=ExchangeType.TOPIC,
    )
)
async def base_handler(
    body: str,
    level: str = Path(),
):
    ...