An Effect[A, E, R] describes a computation. Building one runs nothing; a run function
(run_sync, run_async, run_main) executes it on a fiber.
import pyfect as fx
# An effect is a description of a program. Building one runs nothing.
one = fx.succeed(1)
now = fx.sync(lambda: 41) # runs the function each time the effect runs
answer = now.map(lambda n: n + 1)
both = one.flat_map(lambda a: answer.map(lambda b: a + b))
assert fx.run_sync(answer) == 42
assert fx.run_sync(both) == 43fx.succeed(value)wraps a value;fx.sync(f)callsfeach time the effect runs.maptransforms the value;flat_mapcontinues with another effect.
Generators#
Chains of flat_map get hard to read. Write the effect as a generator instead: yield from
an effect to get its value. @fx.gen makes an effect from a generator; @fx.fn makes a
function that returns one.
import pyfect as fx
@fx.gen
def program():
a = yield from fx.succeed(20)
b = yield from fx.sync(lambda: 22)
yield from fx.sleep(0.01) # suspends this fiber, not the thread
return a + b
# program is an Effect[int]: the type of the return value is inferred.
assert fx.run_sync(program) == 42
@fx.fn
def add(a: int, b: int):
total = yield from fx.succeed(a + b)
return total
# A function returning an effect; add(1, 2) is an Effect[int].
assert fx.run_sync(add(1, 2)) == 3pyright infers the success type from return, the errors from every effect that can fail and
the services from every service used. You rarely write Effect[...] yourself, except on
function signatures you want to pin.
Running effects#
| Function | Use it for |
|---|---|
run_sync(eff) | scripts and tests: no event loop needed, sleeps and forks work |
run_sync_exit(eff) | the same, returning an Exit (Success or Failure) instead of raising |
await run_async(eff) | inside asyncio code |
run_main(eff) | a program's entry point: handles SIGINT/SIGTERM, logs failures, sets the exit code |
ManagedRuntime.make(layer) | building services once and running many effects with them (web servers) |
The run functions accept only effects whose error and service types are Never, so an unhandled
error or a missing service is a type error at the call site.
import asyncio
import pyfect as fx
async def fetch() -> str:
await asyncio.sleep(0.01)
return "data"
# from_async wraps a coroutine function; it needs an event loop to run on.
program = fx.from_async(fetch).map(str.upper)
# run_async runs on the current asyncio loop.
assert asyncio.run(fx.run_async(program)) == "DATA"
# Effects that don't touch asyncio also run without a loop.
assert fx.run_sync(fx.sleep(0.01).as_("done")) == "done"
# run_sync_exit returns an Exit instead of raising.
assert fx.run_sync_exit(fx.succeed(1)) == fx.Success(1)fx.from_async and fx.try_async wrap coroutines; fx.blocking and fx.try_blocking move
blocking calls to a thread.
See the reference for Effect (its methods) and
pyfect.effect (constructors).