From 70592c37afad448751ef1eaf466dbcb87a0ea5df Mon Sep 17 00:00:00 2001 From: Jonathan Sick Date: Thu, 22 Feb 2024 18:46:29 -0500 Subject: [PATCH] try to use KafkaMessage --- src/unfurlbot/handlers/kafka.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/src/unfurlbot/handlers/kafka.py b/src/unfurlbot/handlers/kafka.py index e2ee869..3a6185e 100644 --- a/src/unfurlbot/handlers/kafka.py +++ b/src/unfurlbot/handlers/kafka.py @@ -2,7 +2,7 @@ from __future__ import annotations -from faststream.kafka.fastapi import Context, KafkaRouter +from faststream.kafka.fastapi import Context, KafkaMessage, KafkaRouter from faststream.security import BaseSecurity from pydantic import BaseModel from rubin.squarebot.models.kafka import SquarebotSlackMessageValue @@ -33,10 +33,17 @@ class SlackMessageContainer(BaseModel): group_id=config.consumer_group_id, ) async def handle_slack_message( - msg: str, + body: str, + msg: str = KafkaMessage, topic: str = Context("message.topic"), ) -> None: """Handle a Slack message.""" logger = get_logger(__name__) - message = SquarebotSlackMessageValue.model_validate_json(msg) - logger.info("Slack message text", text=message.text, topic=topic) + message = SquarebotSlackMessageValue.model_validate_json(body) + logger.info( + "Slack message text", + text=message.text, + topic=topic, + msg=msg, + msg_topic=msg.topic, + )