Skip to content

rigging.util

await_(*coros: t.Coroutine[t.Any, t.Any, R]) -> R | list[R] #

A utility function that allows awaiting coroutines in a managed thread.

Parameters:

  • *coros (Coroutine[Any, Any, R], default: () ) –

    Variable number of coroutines to await.

Returns:

  • R | list[R] –

    A single result if one coroutine is passed or a list of results if multiple coroutines are passed.

Source code in rigging/util.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def await_(*coros: t.Coroutine[t.Any, t.Any, R]) -> R | list[R]:  # type: ignore [misc]
    """
    A utility function that allows awaiting coroutines in a managed thread.

    Args:
        *coros: Variable number of coroutines to await.

    Returns:
        A single result if one coroutine is passed or a list of results if multiple coroutines are passed.
    """
    loop = _get_event_loop()
    tasks = [asyncio.run_coroutine_threadsafe(coro, loop) for coro in coros]
    results = [task.result() for task in tasks]
    if len(coros) == 1:
        return results[0]
    return results

escape_xml(xml_string: str) -> str #

Escape XML special characters in a string.

Source code in rigging/util.py 
69
70
71
72
73
def escape_xml(xml_string: str) -> str:
    """Escape XML special characters in a string."""
    prepared = re.sub(r"&(?!(?:amp|lt|gt|apos|quot);)", "&", xml_string)

    return prepared

unescape_xml(xml_string: str) -> str #

Unescape XML special characters in a string.

Source code in rigging/util.py 
76
77
78
79
80
81
82
83
84
def unescape_xml(xml_string: str) -> str:
    """Unescape XML special characters in a string."""
    unescaped = re.sub(r"&", "&", xml_string)
    unescaped = re.sub(r"&lt;", "<", unescaped)
    unescaped = re.sub(r"&gt;", ">", unescaped)
    unescaped = re.sub(r"&apos;", "'", unescaped)
    unescaped = re.sub(r"&quot;", '"', unescaped)

    return unescaped