ohneio.py 文件源码

def protocol(func: typing.Callable[..., ProtocolGenerator[R]]) -> typing.Callable[..., Consumer[R]]:
    """Wraps a Ohne I/O protocol function.

    Under the hood this wraps the generator inside a :class:`~ohneio.Consumer`.

    Args:
        func (callable): Protocol function to wrap. (Protocol functions have to be generators)

    Returns:
        callable: wrapped function.
    """
    if not callable(func):  # pragma: no cover
        # This is for users misusing the library, type hinting already checks this
        raise ValueError("A protocol needs to a be a callable")
    if not inspect.isgeneratorfunction(func):  # pragma: no cover
        # This is for users misusing the library, type hinting already checks this
        raise ValueError("A protocol needs to be a generator function")

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return Consumer(func(*args, **kwargs))

    return wrapper