Skip to content

timeout_scope

faststream.utils.functions.timeout_scope #

timeout_scope(timeout=30, raise_timeout=False)
Source code in faststream/utils/functions.py 
58
59
60
61
62
63
64
65
def timeout_scope(
    timeout: Optional[float] = 30,
    raise_timeout: bool = False,
) -> ContextManager[anyio.CancelScope]:
    scope: Callable[[Optional[float]], ContextManager[anyio.CancelScope]]
    scope = anyio.fail_after if raise_timeout else anyio.move_on_after  # type: ignore[assignment]

    return scope(timeout)