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)
|