Skip to content

MQTTSubscriberSpecification

faststream.mqtt.subscriber.specification.MQTTSubscriberSpecification #

MQTTSubscriberSpecification(
    _outer_config: MQTTBrokerConfig,
    specification_config: MQTTSubscriberSpecificationConfig,
    calls: CallsCollection[Any],
)

Bases: SubscriberSpecification[MQTTBrokerConfig, MQTTSubscriberSpecificationConfig]

Source code in faststream/mqtt/subscriber/specification.py 
24
25
26
27
28
29
30
def __init__(
    self,
    _outer_config: "MQTTBrokerConfig",
    specification_config: "MQTTSubscriberSpecificationConfig",
    calls: "CallsCollection[Any]",
) -> None:
    super().__init__(_outer_config, specification_config, calls)

address property #

address: str

The topic a message actually arrives on.

A shared subscription is asked for by prefixing the topic with $share/<group>/, but no message ever carries that prefix in its topic, so the address does not either. The channel name and the MQTT channel binding both keep it, which is where the group stays visible.

topic property #

topic: str

channel_labels property #

channel_labels: list[str]

config instance-attribute #

config = specification_config

include_in_schema property #

include_in_schema: bool

calls instance-attribute #

calls = calls

description property #

description: str | None

call_name property #

call_name: str

name property #

name: str

The key of the first channel this endpoint names.

get_schema #

get_schema() -> dict[str, SubscriberSpec]
Source code in faststream/mqtt/subscriber/specification.py 
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
def get_schema(self) -> dict[str, SubscriberSpec]:
    payloads = self.get_payloads()

    return {
        self.name: SubscriberSpec(
            address=self.address,
            description=self.description,
            operation=Operation(
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(payloads),
                ),
                bindings=OperationBinding(
                    mqtt=mqtt_bindings.OperationBinding(
                        qos=self.config.qos,
                    ),
                ),
            ),
            bindings=ChannelBinding(
                mqtt=mqtt_bindings.ChannelBinding(
                    topic=self.topic,
                    qos=self.config.qos,
                ),
            ),
        ),
    }

get_payloads #

get_payloads() -> list[tuple[dict[str, Any], str]]
Source code in faststream/_internal/endpoint/subscriber/specification.py 
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
def get_payloads(self) -> list[tuple["dict[str, Any]", str]]:
    payloads: list[tuple[dict[str, Any], str]] = []

    call_name = self.call_name

    for h in self.calls:
        if h.dependant is None:
            msg = "You should setup `Handler` at first."
            raise SetupError(msg)

        body = parse_handler_params(
            h.dependant,
            prefix=f"{self.config.title_ or call_name}:Message",
        )
        payloads.append((body, to_camelcase(h.name)))

    if not self.calls:
        payloads.append(
            (
                {
                    "title": f"{self.config.title_ or call_name}:Message:Payload",
                },
                to_camelcase(call_name),
            ),
        )

    return payloads