handle incoming message validation error

[pablo-esteban]

Hi, I want to handle the validation error raised by pydantic if the incoming json does not conform to the expected type. How do I do this?

i was hoping to specify something like an exception handler (using the example from the docs):

from pydantic import BaseModel, Field, NonNegativeInt


from faststream import FastStream
from faststream.kafka import KafkaBroker

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


class UserInfo(BaseModel):
    name: str = Field(
        ..., examples=["John"], description="Registered user name"
    )
    user_id: NonNegativeInt = Field(
        ..., examples=[1], description="Registered user id"
    )


def on_validation_err(exception, value):
     log = logging.getLogger()
     log.error(f"Could not validate {value}")
     # maybe raise a different exception


@broker.subscriber("test-topic", on_error={pydantic_core._pydantic_core.ValidationError: on_validation_err})
async def handle(
    user: UserInfo,
):
    assert user.name == "John"
    assert user.user_id == 1

[yann-combarnous]

You can use something like:

from faststream import ExceptionMiddleware

async def error_handler(
        self, exc: Exception, message: KafkaMessage, context: ContextRepo, broker: BrokerAnnotation, logger: Logger
    ):
        <your code here>

Broker(
...
middlewares=(ExceptionMiddleware(handlers={Exception: error_handler}),)
...
)

[pablo-esteban]

Nice! that looks great.

I went for something like this in the end (as the app grows i should use your approach):

broker = KafkaBroker(
    ...,
    serializer=None,
)


@broker.subscriber("test-topic")
async def on_input_data(msg: dict[str, Any]):
    try:
        obj = UserInfo.model_validate(msg)
    except pydantic.ValidationError as err:
        log.error(err)
        return
    # interesting business logic
    ...