do.py 文件源码

def untyped_do(f: Callable[..., Generator[G, B, None]]) -> Callable[..., G]:
    @functools.wraps(f)
    def do_loop(*a: Any, **kw: Any) -> F[B]:
        itr = f(*a, **kw)
        if not isinstance(itr, GeneratorType):
            raise Exception(f'function `{f.__qualname__}` decorated with `do` does not produce a generator')
        init = itr.send(None)
        m = Monad.fatal_for(init)
        @functools.wraps(f)
        def loop(val: B) -> F[B]:
            try:
                return m.flat_map(itr.send(val), loop)
            except StopIteration:
                return m.pure(val)
        return m.flat_map(init, loop)
    return do_loop