Skip to content

settings_provider_factory

faststream.nats.prometheus.provider.settings_provider_factory #

settings_provider_factory(
    msg: Union[Msg, Sequence[Msg], None],
) -> (
    NatsMetricsSettingsProvider
    | BatchNatsMetricsSettingsProvider
    | None
)
Source code in faststream/nats/prometheus/provider.py 
54
55
56
57
58
59
60
61
62
def settings_provider_factory(
    msg: Union["Msg", Sequence["Msg"], None],
) -> NatsMetricsSettingsProvider | BatchNatsMetricsSettingsProvider | None:
    if isinstance(msg, Sequence):
        return BatchNatsMetricsSettingsProvider()
    if isinstance(msg, Msg) or msg is None:
        return NatsMetricsSettingsProvider()
    # KeyValue and Object Storage watch cases
    return None