Is it possible to get the message key when using the fastapi plugin?

[iamvishnuks]

Hi I have a bare minimum code as below,

from fastapi import FastAPI, APIRouter
from fastapi.middleware.cors import CORSMiddleware
from faststream.security import SASLPlaintext
from faststream.confluent.fastapi import KafkaRouter
from faststream.confluent.message import KafkaMessage
from faststream.confluent.fastapi import Context
from typing import Annotated



security = SASLPlaintext(username="username", password="passwor", use_ssl=True)
kafka_router = KafkaRouter("test:9092", security=security, client_id="testclient")

app = FastAPI(lifespan=kafka_router.lifespan_context)

origins = [
    '*',
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)


@kafka_router.subscriber("dev_topic", group_id="testclient")
async def kafka_handler(msg: dict ):
    print(msg)

When I print the msg I am able to parse the message value which was in json format, but how could I get the message key from it. I couldn't find any examples in fastapi plugins docs. Can anyone please help. Thanks!!

[Lancetnik]

You can use faststream.confluen.fastapi.Context the same way with the regular one: https://faststream.airt.ai/latest/confluent/message/#message-fields-access

So, you should get it by smth like this:

from faststream.confluent.fastapi import Context

@router.subscriber("test")
async def base_handler(
    body: str,
    key: str = Context("message.raw_message.key"),
):
    print(key)

[iamvishnuks]

Thanks !! above code did work with a small change in the print statement
print(headers())

[Lancetnik]

Sure, I am just copied from the doc with a little change