Annotation Serialization#

Basic usage#

As you already know, FastStream serializes your incoming message body according to the function type annotations using Pydantic.

So, there are some valid use cases:

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

@broker.subscriber("test")
async def handle(
    msg: str,
):
    ...

@broker.subscriber("test")
async def handle(
    msg: bytes,
):
    ...

@broker.subscriber("test")
async def handle(
    msg: int,
):
    ...

The same applies to other Python primitive types as well (float, bool, datetime, etc.)

Also, thanks to Pydantic (again), FastStream is able to serialize (and validate) more complex types like pydantic.HttpUrl, pydantic.PositiveInt, etc.

JSON Basic Serialization#

But how can we serialize a more complex message, like { "name": "John", "user_id": 1 } ?

For sure, we can serialize it as a simple dict

1
2
3
4
5
6
7

from typing import Any

@broker.subscriber("test")
async def handle(
    msg: dict[str, Any],
):
    ...

But it doesn't look like a correct message validation, does it?

For this reason, FastStream supports per-argument message serialization: you can declare multiple arguments with various types and your message will be unpacked into them:

1
2
3
4
5
6
7

@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

1
2
3
4
5
6
7

@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

1
2
3
4
5
6
7

@broker.subscriber("test-queue")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

1
2
3
4
5
6
7

@broker.subscriber("test-subject")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

1
2
3
4
5
6
7

@broker.subscriber("test-channel")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

1
2
3
4
5
6
7

@broker.subscriber("test-topic")
async def handle(
    name: str,
    user_id: int,
):
    assert name == "John"
    assert user_id == 1

Serialization details#

Simple message#

If you expect to consume a simple message like b"1" or b"any_string", use a single argument as the function annotation.

In this case, your argument name does not matter, because it is the whole message body.

See the examples below:

1
2

async def handler(body: int): ...
    # waits any int-serializable simple message like b"1"

1
2

async def handler(body: str): ...
    # waits any str-serializable simple message like b"any_string"

JSON-like message#

If you expect to consume a message with a specific structure like JSON, multiple arguments are a shortcut for JSON.

In this case, your message will be unpacked and serialized by its fields.

See the examples below:

1
2

async def handler(name: str, id: int): ...
    # waits for { "name": "John", "id": 1, ... }

To consume a single JSON object, you should create a single-field pydantic model and use it for annotation.

1
2
3
4
5

class User(BaseModel):
    name: str

async def handler(body: User): ...
    # waits for { "name": "John" }

Partial body consuming#

If you don't need to use all the fields, you can simply specify the fields you want to use, and the others will be ignored. See the example below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker()
app = FastStream(broker)

@broker.subscriber("test")
async def handle(name: str, age: int):
    print(f"{name=}, {age=}")

@app.after_startup
async def t():
    await broker.publish({
        "name": "John",
        "age": 25,
        "useless": {
            "nested": "useless"
        }
    }, topic="test")