Skip to content

compile_mqtt_path

faststream.mqtt.path.compile_mqtt_path #

compile_mqtt_path(
    path: str,
) -> tuple[Pattern[str] | None, str]

Compile an MQTT topic template with named single-level captures.

{name} captures one complete MQTT topic level and subscribes with the native + wildcard. Multi-level # captures are intentionally not supported; use MQTTMessage.raw_message.topic when the full topic is needed.

Source code in faststream/mqtt/path.py
def compile_mqtt_path(path: str) -> tuple[Pattern[str] | None, str]:
    """Compile an MQTT topic template with named single-level captures.

    ``{name}`` captures one complete MQTT topic level and subscribes with the
    native ``+`` wildcard. Multi-level ``#`` captures are intentionally not
    supported; use ``MQTTMessage.raw_message.topic`` when the full topic is
    needed.
    """
    escaped_path = _escape_literal_braces(path)

    for match in MQTT_PARAM_REGEX.finditer(escaped_path):
        name = match.group(1)
        start, end = match.start(), match.end()
        before = escaped_path[start - 1] if start > 0 else ""
        after = escaped_path[end] if end < len(escaped_path) else ""

        if before not in MQTT_TOPIC_BOUNDARIES or after not in MQTT_TOPIC_BOUNDARIES:
            msg = (
                f"Param {{{name}}} must occupy a whole topic level "
                f"(surrounded by '/' or string boundaries) in topic {path!r}"
            )
            raise SetupError(msg)

    path_regex, mqtt_topic = compile_path(
        escaped_path,
        replace_symbol="+",
        patch_regex=_patch_mqtt_regex,
        param_regex="[^/]+",
    )
    return path_regex, _restore_literal_braces(mqtt_topic)