I'll show how to represent functional effects in Python and how we can implement return and >>= functions for Python coroutines. Then we'll formulate Monad laws using these and finally implement a simple working example of pure functional Python program with synchronous and asynchronous effects.

I assume Python's asyncio is monadic the same way Javascript's Promise is. My original motivation was to check what would happen if any of the monad laws didn't hold for it. Unfortunately, it turns out these three laws feel so fundamental that I couldn't really think of a way to hypothetically break them. So instead, I'll just show a working example of formulation of these laws in Python for asyncio along with usual representation of IO.

Monad laws

Let's summarise monad laws using Haskell pseudo code (taken from the haskell wiki). Think of a being a pure value (for example string or number) and h a function taking a and returning for example a coroutine.

• return is a function that takes a pure value A and creates a value wrapped in the monad. For example, we can have a str and create Coroutine[Any, Any, str] using such a function. We will call it of later.
• >>= is a function that takes value A in a monad and a function A -> m B (for example str -> Coroutine[Any, Any, int]) and returns m B (Coroutine[Any, Any, int]). We also call it chain if we switch the order of arguments.

Left identity

return a >>= h     ===     h a

This law practically says it doesn't matter whether

• we put a pure value into a coroutine and then chain a function h on it, or
• we invoke h with a directly.

Right identity

m >>= return       ===     m

This one is even easier. It says that chaining a function that puts its input into the context doesn't have any effect on it. It is somewhat a boosted version of applying an identity function.

Associativity

(m >>= g) >>= h    ===     m >>= (\x -> g x >>= h)

The formulation of associativity using >>= doesn't look like an analogous associativity of function: f(g(x)) === (f . g)(x), but it actually says the same. For effects it means that doing fa := m >>= g and then fa >>= h is the same as composing g and h into new gh and doing m >>= gh.

Implementing return and >>= in Python

Let's introduce some new types to simplify the formulation in Python. Firstly, we'll create a type Fn[A, B] to denote functions with a single argument A returning B. For example

• print which accepts an object convertible to str and returns None,
• def add1(a: int) -> int: return a + 1 which takes an int and returns another int

Then we'll have two types representing functional effects.

• Task[A] representing an asynchronous always succeeding effect returning value of type A
• IO[A] representing a synchronous always succeeding effect returning value of type A

The intuition behind Task[A] is very easy because it is an alias to actual Python's coroutine, an evaluation of value of type Task[A] doesn't really trigger any effects. We need to run such an object in the event loop and only then all the side-effects defined in the coroutine actually happen.

With the same mindset we also introduce IO[A] which is declared but not-yet-evaluated synchronous effect. It is much simpler because we don't need a special runner as we do in case of the asyncio and evaluation of the IO effect can be simply done by invoking the function (we declared IO[A] to be a function with zero arguments returning A).

from typing import Any, Protocol, TypeVar, Callable, Coroutine

A = TypeVar("A")
B = TypeVar("B")

Fn = Callable[[A], B]

Task = Coroutine[Any, Any, A]
IO = Callable[[], A]

Now, we will create the return function and name it of instead. Its only purpose is to create a coroutine that contains its input value.

# of :: A -> Task[A]
async def of(a: A) -> A:
    return a

We will create the chain function which behaves the same as >>= but arguments are in opposite order.

# chain :: (A -> Task[B]) -> Task[A] -> Task[B]
async def chain(fn: Fn[A, Task[B]], fa: Task[A]) -> B:
    return await fn(await fa)

Now in case of coroutines, if f1 and f2 are functions returning coroutines, then running result = chain(f1, f2(input)) is equivalent to

intermediate_result = await f2(input)
result = await f1(intermediate_result)

Monad laws in Python

Having chain and of, it is now rather straightforward to rewrite the Haskell formulation into (pseudo) Python.

chain(h, of(a)) === h(a)
chain(of, m) === m
chain(h, chain(g, m)) === chain(lambda x: chain(h, g(x)), m)

Combining IO and Task in Python

Below, we can see a working example of a purely functional Python program run in the event loop that combines blocking and asynchronous effects. I added another chain_io combinator which allows us to run IO effect in the Task context.

The program doesn't really do much, it simulates a long-running asynchronous operation (in the do_stuff function) whose result is printed to the standard output.

import asyncio
from typing import Any, Protocol, TypeVar, Callable, Coroutine

A = TypeVar("A")
B = TypeVar("B")

Task = Coroutine[Any, Any, A]
IO = Callable[[], A]
Fn = Callable[[A], B]

async def of(a: A) -> A:
    return a

async def chain(fn: Fn[A, Task[B]], fa: Task[A]) -> B:
    return await fn(await fa)

async def chain_io(fn: Fn[A, IO[B]], fa: Task[A]) -> B:
    return fn(await fa)()

# interop with the inpure world

class ToString(Protocol):
    def __str__(self) -> str:
        ...

def print_io(message: ToString) -> IO[None]:
    return lambda: print(message)

def print_task(fa: Task[ToString]) -> Task[None]:
    return chain_io(print_io, fa)

# program

async def do_stuff(name: str) -> tuple[str, int]:
    await asyncio.sleep(1)
    return (f"The most beautiful name in the world is {name}", 69)

program = print_task(chain(do_stuff, of("Milan")))

asyncio.run(program)