See the nesting problem with plain lists

nums = [1, 2, 3]
both = lambda x: [x, -x]      # a step that returns a box

print(list(map(both, nums)))  # map: boxes inside a box
print([y for x in nums for y in both(x)])  # bind: flat

[[1, -1], [2, -2], [3, -3]]
[1, -1, 2, -2, 3, -3]

map keeps the extra layer. After three steps you would have boxes in boxes in boxes. bind is map followed by flatten, so the shape stays F B no matter how long the chain.

Hold on to this: monad = a box type whose steps can be chained without nesting. The next slides make it exact.

You already use functors

list(map(str.upper, ["a", "b"]))   # list functor
# ['A', 'B']

{k: v * 2 for k, v in d.items()}    # dict values

# Optional as a functor, by hand:
def fmap_opt(f, x):
    return None if x is None else f(x)

The limit: map cannot handle an f that itself returns a box. You get F(F B). Monads fix that.

Picture it

Minimal Python interface: a static unit(x) and a method bind(self, f). map comes for free: m.bind(lambda x: unit(f(x))).

Test them like any property

f = lambda x: Just(x + 1)
g = lambda x: Just(x * 10) if x < 100 else Nothing()
for a in [1, 99, 150]:
    m = Just(a)
    assert Maybe.unit(a).bind(f) == f(a)
    assert m.bind(Maybe.unit) == m
    assert m.bind(f).bind(g) == m.bind(lambda x: f(x).bind(g))
print("laws ok")

from dataclasses import dataclass

class Maybe:
    @staticmethod
    def unit(x): return Just(x)

@dataclass(frozen=True)
class Just(Maybe):
    value: object
    def map(self, f):  return Just(f(self.value))
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Nothing(Maybe):
    def map(self, f):  return self
    def bind(self, f): return self   # stop here

USERS = {1: {"name": "Ada", "address_id": 10},
         2: {"name": "Bob"}}
ADDRS = {10: {"city": "London", "zip": "N1 9GU"}}

def get(d, k):
    return Just(d[k]) if k in d else Nothing()

def zip_for(uid):
    return (get(USERS, uid)
            .bind(lambda u: get(u, "address_id"))
            .bind(lambda a: get(ADDRS, a))
            .bind(lambda addr: get(addr, "zip")))

print(zip_for(1), zip_for(2), zip_for(9))

Just(value='N1 9GU') Nothing() Nothing()

Math view: Maybe A = 1 + A (a disjoint sum). unit is the right injection.

@dataclass(frozen=True)
class Ok:
    value: object
    def bind(self, f): return f(self.value)

@dataclass(frozen=True)
class Err:
    error: str
    def bind(self, f): return self

def parse_int(s):
    try: return Ok(int(s))
    except ValueError: return Err(f"not a number: {s!r}")
def positive(n):
    return Ok(n) if n > 0 else Err(f"must be > 0, got {n}")
def safe_recip(n):
    return Ok(1 / n)

for s in ["4", "-2", "abc"]:
    print(Ok(s).bind(parse_int).bind(positive).bind(safe_recip))

Ok(value=0.25)
Err(error='must be > 0, got -2')
Err(error="not a number: 'abc'")

class ListM:
    def __init__(self, xs): self.xs = list(xs)
    @staticmethod
    def unit(x): return ListM([x])
    def bind(self, f):
        return ListM(y for x in self.xs for y in f(x).xs)

guard = lambda ok: ListM([None]) if ok else ListM([])

def triples(n):
    return ListM(range(1, n)).bind(lambda a:
           ListM(range(a, n)).bind(lambda b:
           ListM(range(b, n)).bind(lambda c:
           guard(a*a + b*b == c*c).bind(lambda _:
           ListM.unit((a, b, c))))))

print(triples(20).xs)

[(3, 4, 5), (5, 12, 13), (6, 8, 10), (8, 15, 17), (9, 12, 15)]

@dataclass(frozen=True)
class Writer:
    value: object
    log: tuple = ()
    @staticmethod
    def unit(x): return Writer(x)          # empty log
    def bind(self, f):
        w = f(self.value)
        return Writer(w.value, self.log + w.log)

def double(x): return Writer(x * 2, (f"doubled {x}",))
def inc(x):    return Writer(x + 1, (f"incremented {x}",))

print(Writer.unit(5).bind(double).bind(inc))

Writer(value=11, log=('doubled 5', 'incremented 10'))

from typing import Callable

@dataclass(frozen=True)
class Reader:
    run: Callable                    # env -> value
    @staticmethod
    def unit(x): return Reader(lambda env: x)
    def bind(self, f):
        return Reader(lambda env: f(self.run(env)).run(env))

ask = Reader(lambda env: env)        # read the config

def greeting(name):
    return ask.bind(lambda cfg:
           Reader.unit(f"{cfg['greet']}, {name}!"))

def page(name):
    return greeting(name).bind(lambda g:
           ask.bind(lambda cfg:
           Reader.unit(f"<{cfg['tag']}>{g}</{cfg['tag']}>")))

print(page("Ada").run({"greet": "Hello", "tag": "h1"}))
print(page("Ada").run({"greet": "Hola",  "tag": "p"}))

<h1>Hello, Ada!</h1>
<p>Hola, Ada!</p>

@dataclass(frozen=True)
class State:
    run: Callable              # s -> (value, new_s)
    @staticmethod
    def unit(x): return State(lambda s: (x, s))
    def bind(self, f):
        def step(s):
            a, s2 = self.run(s)
            return f(a).run(s2)  # pass new state along
        return State(step)

get_s = State(lambda s: (s, s))
def put(n): return State(lambda s: (None, n))

def rand():                    # a tiny LCG
    return get_s.bind(lambda seed:
           put((1103515245 * seed + 12345) % 2**31).bind(lambda _:
           State.unit(seed % 100)))

three = rand().bind(lambda a: rand().bind(lambda b:
        rand().bind(lambda c: State.unit([a, b, c]))))
print(three.run(42))
print(three.run(42))           # same seed, same result

([42, 27, 64], 1000676753)
([42, 27, 64], 1000676753)

def do(unit):
    def deco(genfn):
        def run(*args):
            gen = genfn(*args)
            def step(value):
                try:
                    m = gen.send(value)     # next monadic value
                except StopIteration as done:
                    return unit(done.value) # 'return x' -> unit(x)
                return m.bind(step)         # continue after bind
            return step(None)
        return run
    return deco

@do(Maybe.unit)
def zip_for(uid):
    user = yield get(USERS, uid)
    aid  = yield get(user, "address_id")
    addr = yield get(ADDRS, aid)
    return addr["zip"]

print(zip_for(1), zip_for(2))

Just(value='N1 9GU') Nothing()

def join(mm):                 # mu
    return mm.bind(lambda m: m)

def fmap(f, m, unit):         # T f
    return m.bind(lambda x: unit(f(x)))

def bind_via_join(m, f, unit):
    return join(fmap(f, m, unit))

print(join(Just(Just(3))), join(Just(Nothing())))

Just(value=3) Nothing()

So what?

The three odd-looking laws are just identity + associativity for composing effectful functions. A monad is precisely what makes A → T B functions compose like normal ones.

def kleisli(f, g):            # f >=> g
    return lambda x: f(x).bind(g)

pipeline = kleisli(kleisli(parse_int, positive), safe_recip)
print(pipeline("8"), pipeline("0"))

Ok(value=0.125) Err(error='must be > 0, got 0')

Moggi's insight: a program A → B with effects is a Kleisli arrow A → T B. Pick T to pick the effect.

Applicative example: validate all fields

def validate_all(*results):
    errs = [r.error for r in results if isinstance(r, Err)]
    return Err("; ".join(errs)) if errs else Ok(tuple(r.value for r in results))

print(validate_all(parse_int("x"), parse_int("7"), positive(-1)))

Err(error="not a number: 'x'; must be > 0, got -1")

An error-accumulating Validation applicative differs from the fail-fast applicative induced by Result's monadic bind. Choose the intended semantics explicitly.

Libraries

returns (dry-python) gives typed Maybe, Result, IO, and RequiresContext with mypy support. parsy and pyparsing use monadic parsing.