Stanislav Hejny
38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
import asyncio
|
|
import re
|
|
|
|
pattern = re.compile(r"[^a-zA-Z0-9.#/\-]")
|
|
AWAIT_SLEEP = 0.05 # seconds
|
|
|
|
|
|
def sanitize_as_rabbitmq_name(input_str: str) -> str:
|
|
matcher = pattern.search(input_str)
|
|
token = input_str[: matcher.start()] if matcher else input_str
|
|
return sanitize_as_rabbitmq_domain_name(token)
|
|
|
|
|
|
def sanitize_as_rabbitmq_domain_name(token: str) -> str:
|
|
# Step 1: Replace non-alphanumeric characters (except periods) with a period
|
|
step1 = re.sub(r"[^A-Za-z0-9.#]", ".", token)
|
|
# Step 2: Replace multiple consecutive periods with a single period
|
|
step2 = re.sub(r"\.+", ".", step1)
|
|
# Step 3: Remove trailing '.' character, if applicable
|
|
step_last_char = len(step2) - 1
|
|
return step2[:-1] if step_last_char > 0 and step2[step_last_char] == "." else step2
|
|
|
|
|
|
async def await_future(future: asyncio.Future, sleep_time=AWAIT_SLEEP):
|
|
"""
|
|
Wait for a future to complete.
|
|
"""
|
|
while not future.done():
|
|
await asyncio.sleep(sleep_time)
|
|
|
|
|
|
async def await_result(future: asyncio.Future, sleep_time=AWAIT_SLEEP):
|
|
"""
|
|
Wait for a future to complete and return its result.
|
|
"""
|
|
await await_future(future, sleep_time=sleep_time)
|
|
return future.result()
|