Files
full-stack-fastapi/backend/app/mqtt/topic.py

11 lines
289 B
Python

import re
TOPIC_PATTERN = re.compile(r"^terminal/(?P<device_id>[^/]+)/raw$")
def extract_device_id(topic: str) -> str:
match = TOPIC_PATTERN.match(topic)
if match is None:
raise ValueError("topic must match terminal/{device_id}/raw")
return match.group("device_id")