Bases: BaseKafkaMetricsSettingsProvider['ConsumerRecord']
Source code in faststream/kafka/prometheus/provider.py
| def __init__(self) -> None:
self.messaging_system = "kafka"
|
messaging_system instance-attribute
messaging_system = 'kafka'
get_publish_destination_name_from_kwargs
get_publish_destination_name_from_kwargs(kwargs)
Source code in faststream/kafka/prometheus/provider.py
| def get_publish_destination_name_from_kwargs(
self,
kwargs: "AnyDict",
) -> str:
return cast(str, kwargs["topic"])
|
get_consume_attrs_from_message
get_consume_attrs_from_message(msg)
Source code in faststream/kafka/prometheus/provider.py
| def get_consume_attrs_from_message(
self,
msg: "StreamMessage[ConsumerRecord]",
) -> "ConsumeAttrs":
return {
"destination_name": msg.raw_message.topic,
"message_size": len(msg.body),
"messages_count": 1,
}
|