runtime

effectpy.runtime

Fiber

Bases: Generic[E, A]

A lightweight async task with structured cancellation.

Fibers are effectpy's unit of concurrent execution. Unlike raw asyncio Tasks, fibers provide structured error handling and proper cancellation semantics.

Parameters:

Name	Type	Description	Default
task	Task	The underlying asyncio task	required
name	Optional[str]	Optional name for debugging	None

Attributes:

Name	Type	Description
id	str	Unique identifier for this fiber
name	Optional[str]	Optional name for debugging
status	str	Current status ('running', 'done', 'failed', 'cancelled')

await_() async

Wait for this fiber to complete and get the structured result.

Returns:

Type	Description
Exit[E, A]	Exit containing either the success value or failure cause

Example

fiber = runtime.fork(my_effect())
exit_result = await fiber.await_()

if exit_result.success:
    print(f"Success: {exit_result.value}")
else:
    print(f"Failure: {exit_result.cause.render()}")

interrupt()

Cancel this fiber.

Sends a cancellation signal to the underlying task. The fiber will transition to 'cancelled' status.

Example

fiber = runtime.fork(long_running_task())
# Later...
fiber.interrupt()  # Cancel the fiber

join() async

Wait for this fiber to complete and get the success value.

Returns:

Type	Description
A	The success value if the fiber succeeds

Raises:

Type	Description
Failure	If the fiber fails with a business logic error
Exception	If the fiber dies with an unexpected exception

Note

This is similar to await_() but throws exceptions instead of returning structured Exit values.

Runtime

Manages effect execution and fiber lifecycle.

Runtime provides advanced features for concurrent execution including fiber management, supervision, and structured error handling.

Parameters:

Name	Type	Description	Default
base	Optional[Context]	Base context for all effects (default: empty Context)	None
supervisor	Optional[Supervisor]	Supervisor for fiber lifecycle callbacks (default: no-op)	None

Example

runtime = Runtime()

# Fork effects as fibers
fiber1 = runtime.fork(fetch_data("source1"))
fiber2 = runtime.fork(fetch_data("source2"))

# Wait for results
result1 = await fiber1.join()
result2 = await fiber2.join()

await runtime.shutdown()

fork(eff, name=None)

Fork an effect as a new fiber.

The effect starts executing immediately in the background.

Parameters:

Name	Type	Description	Default
eff	Effect[Any, E, A]	The effect to execute	required
name	Optional[str]	Optional name for debugging	None

Returns:

Type	Description
Fiber[E, A]	A Fiber representing the running effect

Example

fiber = runtime.fork(long_running_task(), name="background-worker")

# Do other work...

result = await fiber.join()  # Wait for completion