Skip to content

PrometheusMiddleware

faststream.prometheus.middleware.PrometheusMiddleware #

PrometheusMiddleware(
    msg=None, *, settings_provider_factory, metrics_manager
)

Bases: BaseMiddleware

Source code in faststream/prometheus/middleware.py 
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    msg: Optional[Any] = None,
    *,
    settings_provider_factory: Callable[
        [Any], Optional[MetricsSettingsProvider[Any]]
    ],
    metrics_manager: MetricsManager,
) -> None:
    self._metrics_manager = metrics_manager
    self._settings_provider = settings_provider_factory(msg)
    super().__init__(msg)

msg instance-attribute #

msg = msg

consume_scope async #

consume_scope(call_next, msg)
Source code in faststream/prometheus/middleware.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[Any]",
) -> Any:
    if self._settings_provider is None:
        return await call_next(msg)

    messaging_system = self._settings_provider.messaging_system
    consume_attrs = self._settings_provider.get_consume_attrs_from_message(msg)
    destination_name = consume_attrs["destination_name"]

    self._metrics_manager.add_received_message(
        amount=consume_attrs["messages_count"],
        broker=messaging_system,
        handler=destination_name,
    )

    self._metrics_manager.observe_received_messages_size(
        size=consume_attrs["message_size"],
        broker=messaging_system,
        handler=destination_name,
    )

    self._metrics_manager.add_received_message_in_process(
        amount=consume_attrs["messages_count"],
        broker=messaging_system,
        handler=destination_name,
    )

    err: Optional[Exception] = None
    start_time = time.perf_counter()

    try:
        result = await call_next(await self.on_consume(msg))

    except Exception as e:
        err = e

        if not isinstance(err, IgnoredException):
            self._metrics_manager.add_received_processed_message_exception(
                exception_type=type(err).__name__,
                broker=messaging_system,
                handler=destination_name,
            )
        raise

    finally:
        duration = time.perf_counter() - start_time
        self._metrics_manager.observe_received_processed_message_duration(
            duration=duration,
            broker=messaging_system,
            handler=destination_name,
        )

        self._metrics_manager.remove_received_message_in_process(
            amount=consume_attrs["messages_count"],
            broker=messaging_system,
            handler=destination_name,
        )

        status = ProcessingStatus.acked

        if msg.committed or err:
            status = (
                PROCESSING_STATUS_BY_ACK_STATUS.get(msg.committed)  # type: ignore[arg-type]
                or PROCESSING_STATUS_BY_HANDLER_EXCEPTION_MAP.get(type(err))
                or ProcessingStatus.error
            )

        self._metrics_manager.add_received_processed_message(
            amount=consume_attrs["messages_count"],
            status=status,
            broker=messaging_system,
            handler=destination_name,
        )

    return result

publish_scope async #

publish_scope(call_next, msg, *args, **kwargs)
Source code in faststream/prometheus/middleware.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def publish_scope(
    self,
    call_next: "AsyncFunc",
    msg: Any,
    *args: Any,
    **kwargs: Any,
) -> Any:
    if self._settings_provider is None:
        return await call_next(msg, *args, **kwargs)

    destination_name = (
        self._settings_provider.get_publish_destination_name_from_kwargs(kwargs)
    )
    messaging_system = self._settings_provider.messaging_system

    err: Optional[Exception] = None
    start_time = time.perf_counter()

    try:
        result = await call_next(
            await self.on_publish(msg, *args, **kwargs),
            *args,
            **kwargs,
        )

    except Exception as e:
        err = e
        self._metrics_manager.add_published_message_exception(
            exception_type=type(err).__name__,
            broker=messaging_system,
            destination=destination_name,
        )
        raise

    finally:
        duration = time.perf_counter() - start_time

        self._metrics_manager.observe_published_message_duration(
            duration=duration,
            broker=messaging_system,
            destination=destination_name,
        )

        status = PublishingStatus.error if err else PublishingStatus.success
        messages_count = len((msg, *args))

        self._metrics_manager.add_published_message(
            amount=messages_count,
            status=status,
            broker=messaging_system,
            destination=destination_name,
        )

    return result

on_receive async #

on_receive()

Hook to call on message receive.

Source code in faststream/broker/middlewares/base.py 
18
19
20
async def on_receive(self) -> None:
    """Hook to call on message receive."""
    pass

after_processed async #

after_processed(exc_type=None, exc_val=None, exc_tb=None)

Asynchronously called after processing.

Source code in faststream/broker/middlewares/base.py 
22
23
24
25
26
27
28
29
async def after_processed(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exc_tb: Optional["TracebackType"] = None,
) -> Optional[bool]:
    """Asynchronously called after processing."""
    return False

on_consume async #

on_consume(msg)

Asynchronously consumes a message.

Source code in faststream/broker/middlewares/base.py 
44
45
46
47
48
49
async def on_consume(
    self,
    msg: "StreamMessage[Any]",
) -> "StreamMessage[Any]":
    """Asynchronously consumes a message."""
    return msg

after_consume async #

after_consume(err)

A function to handle the result of consuming a resource asynchronously.

Source code in faststream/broker/middlewares/base.py 
51
52
53
54
async def after_consume(self, err: Optional[Exception]) -> None:
    """A function to handle the result of consuming a resource asynchronously."""
    if err is not None:
        raise err

on_publish async #

on_publish(msg, *args, **kwargs)

Asynchronously handle a publish event.

Source code in faststream/broker/middlewares/base.py 
75
76
77
78
79
80
81
82
async def on_publish(
    self,
    msg: Any,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Asynchronously handle a publish event."""
    return msg

after_publish async #

after_publish(err)

Asynchronous function to handle the after publish event.

Source code in faststream/broker/middlewares/base.py 
84
85
86
87
88
89
90
async def after_publish(
    self,
    err: Optional[Exception],
) -> None:
    """Asynchronous function to handle the after publish event."""
    if err is not None:
        raise err