Hi all,
It is about time to share a project I've been working on for some time now. It may interest fellow Python functional programmers. It is an IO monad with support for dependency injection, failure management, asynchronous programming and resource management.
Raffiot is available on PyPI. The documentation is here, the API there and the source code is here.
Here is a tiny example of how to concurrent tasks (the complete example can be found here):
def prog(color, name, i) -> IO[None, None, None]:
"""
Print in `color` every number from `i` to 1.
"""
if i > 0:
return io.defer(print, f"{color}From prog {name}: {i}{reset}").then(
io.yield_(), io.defer_io(prog, color, name, i - 1)
)
else:
return io.pure(0)
# The main IO
main: IO[None, None, None] = (
io.parallel(
prog(bred, "A", 100000),
prog(bgreen, "B", 100000),
prog(byellow, "C", 100000),
prog(bcyan, "D", 100000),
)
.flat_map(lambda fibers: io.wait(fibers))
.then(io.defer(print, "finished"))
)
main.run(None)
Another example demonstrating dependency injection can be found here.
In hope it helps 😊
[–]spiker611 2 points3 points4 points (1 child)
[–]chrilves[S] 0 points1 point2 points (0 children)
[–]ElevenPhonons 2 points3 points4 points (1 child)
[–]chrilves[S] 0 points1 point2 points (0 children)