Algebraic Effects in Python

effects.py

import asyncio
from greenlet import greenlet
from dataclasses import dataclass


class Effect:
    pass


def run_effect(eff):
    return greenlet.getcurrent().parent.switch(eff)


def run_with_effects(func, args, effects):
    processor = greenlet(func)
    res = processor.switch(*args)

    def call_next(*args):
        nonlocal res
        res = processor.switch(*args)

    while isinstance(res, Effect):
        for typ, func in effects.items():
            if isinstance(res, typ):
                call_next(func(res))
    return res


async def run_async_with_effects(func, args, effects):

    processor = greenlet(func)
    res = processor.switch(*args)

    def call_next(*args):
        nonlocal res
        res = processor.switch(*args)

    while isinstance(res, Effect):
        for typ, func2 in effects.items():
            if isinstance(res, typ):
                im = func2(res)
                if im and getattr(im, '__await__'):
                    im = await im
                call_next(im)
    return res



@dataclass
class Log(Effect):
    message: str


@dataclass
class ReadInput(Effect):
    pass


def read_input():
    return run_effect(ReadInput())


def log(*args):
    return run_effect(Log(*args))


def my_calculation():
    input1 = read_input()
    log(f"Recieved {input1}")
    input2 = read_input()
    log(f"Recieved {input1} and {input2}")
    addition = input1 + input2
    log(f"Addition {addition}")
    return addition


def make_sync_read_input(inputs):
    inputs = iter(inputs)
    def effect_handler(_):
        return next(inputs)
    return effect_handler


def make_async_read_input(inputs):
    inputs = iter(inputs)
    async def effect_handler(_):
        await asyncio.sleep(2)
        return next(inputs)
    return effect_handler


def debug_log(log):
    print(f"Debug log: {log.message}")


addition = run_with_effects(my_calculation, [], {
    ReadInput: make_sync_read_input([23, 46]),
    Log: debug_log
})

print(f"Returned {addition}")


addition = asyncio.run(run_async_with_effects(my_calculation, [], {
    ReadInput: make_async_read_input([23, 46]),
    Log: debug_log
}))

print(f"Returned {addition}")