Skip to content

telemetry_attributes_provider_factory

faststream.confluent.opentelemetry.provider.telemetry_attributes_provider_factory #

telemetry_attributes_provider_factory(
    msg: Union[Message, Sequence[Message], None],
) -> (
    ConfluentTelemetrySettingsProvider
    | BatchConfluentTelemetrySettingsProvider
)
Source code in faststream/confluent/opentelemetry/provider.py 
101
102
103
104
105
106
def telemetry_attributes_provider_factory(
    msg: Union["Message", Sequence["Message"], None],
) -> ConfluentTelemetrySettingsProvider | BatchConfluentTelemetrySettingsProvider:
    if isinstance(msg, Sequence):
        return BatchConfluentTelemetrySettingsProvider()
    return ConfluentTelemetrySettingsProvider()