Flows

Reraise

Bases: SimpleContextManager, Generic[E]

Represents context managers that reraise errors of given error_types as error.

Source code in src/funcs/flows.py

@final
@frozen()
class Reraise(SimpleContextManager, Generic[E]):
    """Represents context managers that reraise errors of given `error_types` as `error`."""

    error: E
    """The error to raise."""
    error_types: AnyErrorTypes
    """The error types to reraise."""

    def __enter__(self) -> None:
        pass

    def __exit__(
        self, error_type: Optional[Type[F]], error: Optional[F], traceback: Optional[Traceback]
    ) -> None:
        if error_type is not None and error is not None:
            if is_subclass(error_type, self.error_types):
                raise self.error from error

error: E instance-attribute

The error to raise.

error_types: AnyErrorTypes instance-attribute

The error types to reraise.

ReraiseWith

Bases: SimpleContextManager, Generic[E]

Represents context managers that reraise errors of given error_types as error, which is computed dynamically from the original error.

Source code in src/funcs/flows.py

@final
@frozen()
class ReraiseWith(SimpleContextManager, Generic[E]):
    """Represents context managers that reraise errors of given `error_types` as `error`,
    which is computed dynamically from the original error.
    """

    into: Into[E]
    """The function that computes the error to raise from the original error."""
    error_types: AnyErrorTypes
    """The error types to reraise."""

    def __enter__(self) -> None:
        pass

    def __exit__(
        self, error_type: Optional[Type[F]], error: Optional[F], traceback: Optional[Traceback]
    ) -> None:
        if error_type is not None and error is not None:
            if is_subclass(error_type, self.error_types):
                raise self.into(error) from error

into: Into[E] instance-attribute

The function that computes the error to raise from the original error.

error_types: AnyErrorTypes instance-attribute

The error types to reraise.

Suppress

Represents context managers that suppress errors of given error_types.

Source code in src/funcs/flows.py

@final
@frozen()
class Suppress:
    """Represents context managers that suppress errors of given `error_types`."""

    error_types: AnyErrorTypes
    """The error types to suppress."""

    def __enter__(self) -> None:
        pass

    @overload
    def __exit__(self, error_type: None, error: None, traceback: None) -> Literal[False]: ...

    @overload
    def __exit__(self, error_type: Type[E], error: E, traceback: Traceback) -> bool: ...

    def __exit__(
        self, error_type: Optional[Type[E]], error: Optional[E], traceback: Optional[Traceback]
    ) -> bool:
        return error_type is not None and is_subclass(error_type, self.error_types)

error_types: AnyErrorTypes instance-attribute

The error types to suppress.

PostProcessing

Bases: Generic[R, S]

Represents decorators that post-process results of function calls.

Source code in src/funcs/flows.py

@final
@frozen()
class PostProcessing(Generic[R, S]):
    """Represents decorators that post-process results of function calls."""

    function: Unary[R, S]

    def __call__(self, function: Callable[P, R]) -> Callable[P, S]:
        @wraps(function)
        def wrap(*args: P.args, **kwargs: P.kwargs) -> S:
            return self.function(function(*args, **kwargs))

        return wrap

WrapWith

Represents decorators that wrap function calls with the given context_manager.

Source code in src/funcs/flows.py

@frozen()
class WrapWith:
    """Represents decorators that wrap function calls with the given `context_manager`."""

    context_manager: AnyContextManager
    """The context manager to wrap function calls with."""

    def __call__(self, function: Callable[P, R]) -> Callable[P, R]:
        @wraps(function)
        def wrap(*args: P.args, **kwargs: P.kwargs) -> R:
            with self.context_manager:
                return function(*args, **kwargs)

        return wrap

context_manager: AnyContextManager instance-attribute

The context manager to wrap function calls with.

once(function: Callable[P, R]) -> Callable[P, R]

Wraps the function to be called once, and then the computed result is returned on subsequent calls.

Parameters:

Name	Type	Description	Default
function	Callable[P, R]	The function to wrap.	required

Returns:

Type	Description
Callable[P, R]	The wrapped function.

Source code in src/funcs/flows.py

def once(function: Callable[P, R]) -> Callable[P, R]:
    """Wraps the function to be called once, and then the
    computed result is returned on subsequent calls.

    Arguments:
        function: The function to wrap.

    Returns:
        The wrapped function.
    """
    result: MarkerOr[R] = marker

    @wraps(function)
    def wrap(*args: P.args, **kwargs: P.kwargs) -> R:
        nonlocal result

        if not is_marker(result):
            return result

        result = function(*args, **kwargs)

        return result

    return wrap

reraise(error: E, *error_types: AnyErrorType) -> Reraise[E]

Constructs the Reraise[E] context manager that reraises errors of given error_types as error.

Parameters:

Name	Type	Description	Default
error	E	The error to raise.	required
*error_types	AnyErrorType	The error types to reraise.	()

Returns:

Type	Description
Reraise[E]	The constructed Reraise[E] context manager.

Source code in src/funcs/flows.py

def reraise(error: E, *error_types: AnyErrorType) -> Reraise[E]:
    """Constructs the [`Reraise[E]`][funcs.flows.Reraise] context manager
    that reraises errors of given `error_types` as `error`.

    Arguments:
        error: The error to raise.
        *error_types: The error types to reraise.

    Returns:
        The constructed [`Reraise[E]`][funcs.flows.Reraise] context manager.
    """
    return Reraise(error, error_types)

reraise_with(into: Into[E], *error_types: AnyErrorType) -> ReraiseWith[E]

Constructs the ReraiseWith[E] context manager that reraises errors of given error_types as error, which is computed dynamically from the original error.

Parameters:

Name	Type	Description	Default
into	Into[E]	The function that computes the error to raise from the original error.	required
*error_types	AnyErrorType	The error types to reraise.	()

Returns:

Type	Description
ReraiseWith[E]	The constructed ReraiseWith[E] context manager.

Source code in src/funcs/flows.py

def reraise_with(into: Into[E], *error_types: AnyErrorType) -> ReraiseWith[E]:
    """Constructs the [`ReraiseWith[E]`][funcs.flows.ReraiseWith] context manager
    that reraises errors of given `error_types` as `error`, which is computed dynamically
    from the original error.

    Arguments:
        into: The function that computes the error to raise from the original error.
        *error_types: The error types to reraise.

    Returns:
        The constructed [`ReraiseWith[E]`][funcs.flows.ReraiseWith] context manager.
    """
    return ReraiseWith(into, error_types)

suppress(*error_types: AnyErrorType) -> Suppress

Constructs the Suppress context manager used to suppress errors of given error_types.

Parameters:

Name	Type	Description	Default
*error_types	AnyErrorType	The error types to suppress.	()

Returns:

Type	Description
Suppress	The constructed Suppress context manager.

Source code in src/funcs/flows.py

def suppress(*error_types: AnyErrorType) -> Suppress:
    """Constructs the [`Suppress`][funcs.flows.Suppress] context manager used to suppress
    errors of given `error_types`.

    Arguments:
        *error_types: The error types to suppress.

    Returns:
        The constructed [`Suppress`][funcs.flows.Suppress] context manager.
    """
    return Suppress(error_types)

post_processing(function: Unary[R, S]) -> PostProcessing[R, S]

Constructs the PostProcessing[R, S] decorator which post-processes results of function calls.

Parameters:

Name	Type	Description	Default
function	Unary[R, S]	The post-processing function.	required

Returns:

Type	Description
PostProcessing[R, S]	The constructed PostProcessing[R, S] decorator.

Source code in src/funcs/flows.py

def post_processing(function: Unary[R, S]) -> PostProcessing[R, S]:
    """Constructs the [`PostProcessing[R, S]`][funcs.flows.PostProcessing] decorator
    which post-processes results of function calls.

    Arguments:
        function: The post-processing function.

    Returns:
        The constructed [`PostProcessing[R, S]`][funcs.flows.PostProcessing] decorator.
    """
    return PostProcessing(function)

wrap_with(context_manager: AnyContextManager) -> WrapWith

Constructs the WrapWith decorator that wraps function calls with the given context_manager.

Parameters:

Name	Type	Description	Default
context_manager	AnyContextManager	The context manager to wrap function calls with.	required

Returns:

Type	Description
WrapWith	The constructed WrapWith decorator.

Source code in src/funcs/flows.py

def wrap_with(context_manager: AnyContextManager) -> WrapWith:
    """Constructs the [`WrapWith`][funcs.flows.WrapWith] decorator that wraps function calls
    with the given `context_manager`.

    Arguments:
        context_manager: The context manager to wrap function calls with.

    Returns:
        The constructed [`WrapWith`][funcs.flows.WrapWith] decorator.
    """
    return WrapWith(context_manager)