Core API Reference¶

The effectpy.core module contains the fundamental building blocks of the Effect system:

Effect: The main Effect monad for composable async computations
Cause: Structured error representation with failure composition
Exit: Result wrapper indicating success/failure
Combinators: Functions for composing and transforming effects

Core Types and Functions¶

`effectpy.core` ¶

`Cause` `dataclass` ¶

Bases: Generic[E]

Structured representation of Effect failures.

Cause provides rich error information that can represent: - Business logic failures (fail) - Unexpected exceptions (die) - Cancellation signals (interrupt) - Composed failures (both/then)

Attributes:

Name	Type	Description
`kind`	`str`	Type of failure ('fail', 'die', 'interrupt', 'both', 'then')
`left`	`Optional['Cause[E]']`	Left child cause for composed failures
`right`	`Optional['Cause[E]']`	Right child cause for composed failures
`error`	`Optional[E]`	The business logic error for 'fail' causes
`defect`	`Optional[BaseException]`	The exception for 'die' causes
`annotations`	`list[str]`	List of debugging annotations

`both(l, r)` `staticmethod` ¶

Compose two causes representing concurrent failures.

Parameters:

Name	Type	Description	Default
`l`	`'Cause[E]'`	Left cause	required
`r`	`'Cause[E]'`	Right cause	required

Returns:

Type	Description
`'Cause[E]'`	A new Cause with kind='both' containing both failures

`die(ex)` `staticmethod` ¶

Create a Cause representing an unexpected exception.

Parameters:

Name	Type	Description	Default
`ex`	`BaseException`	The exception that was thrown	required

Returns:

Type	Description
`'Cause[E]'`	A new Cause with kind='die'

`fail(e)` `staticmethod` ¶

Create a Cause representing a business logic failure.

Parameters:

Name	Type	Description	Default
`e`	`E`	The error value	required

Returns:

Type	Description
`'Cause[E]'`	A new Cause with kind='fail'

`interrupt()` `staticmethod` ¶

Create a Cause representing cancellation/interruption.

Returns:

Type	Description
`'Cause[E]'`	A new Cause with kind='interrupt'

`render(indent='', include_traces=True)` ¶

Render this Cause as a human-readable string.

Parameters:

Name	Type	Description	Default
`indent`	`str`	String to use for indentation	`''`
`include_traces`	`bool`	Whether to include stack traces for defects	`True`

Returns:

Type	Description
`str`	Formatted string representation of the cause tree

`then(l, r)` `staticmethod` ¶

Compose two causes representing sequential failures.

Parameters:

Name	Type	Description	Default
`l`	`'Cause[E]'`	First cause	required
`r`	`'Cause[E]'`	Second cause	required

Returns:

Type	Description
`'Cause[E]'`	A new Cause with kind='then' representing sequential failure

`Effect` ¶

Bases: Generic[R, E, A]

The core abstraction for async computations in effectpy.

An Effect[R, E, A] represents a lazy async computation that: - Requires environment R (services from Context) - May fail with error E - Succeeds with value A

Effects are lazy - they describe computations but don't execute until you call ._run(context). This enables powerful composition and optimization.

Class Type Parameters:

Name	Description	Default
`R`	Environment type - what services this effect needs	required
`E`	Error type - how this effect can fail	required
`A`	Success type - what this effect returns on success	required

Example

# Simple effect that always succeeds
simple = succeed(42)

# Effect that may fail
risky = fail("something went wrong")

# Composed effect
composed = succeed(10).map(lambda x: x * 2).flat_map(lambda x: succeed(str(x)))

# Run the effect
result = await composed._run(Context())

`catch_all(f)` ¶

Handle all failures from this effect.

If this effect fails with error E, apply function f to get a recovery Effect. If this effect succeeds, the success value passes through unchanged.

Parameters:

Name	Type	Description	Default
`f`	`Callable[[E], 'Effect[R, E2, A]']`	Function to handle the error and return a recovery effect	required

Returns:

Type	Description
`'Effect[R, E2, A]'`	A new effect with potentially different error type

Example

safe_divide = divide(10, 0).catch_all(
    lambda error: succeed(f"Error handled: {error}")
)

`flat_map(f)` ¶

Chain this effect with another effect-producing function.

If this effect succeeds with value A, apply function f to get a new Effect[R, E, B]. If this effect fails, the error passes through without calling f.

Parameters:

Name	Type	Description	Default
`f`	`Callable[[A], 'Effect[R, E, B]']`	Function that takes the success value and returns a new effect	required

Returns:

Type	Description
`'Effect[R, E, B]'`	A new effect representing the chained computation

Example

def fetch_user(id: int) -> Effect[Any, str, User]: ...
def fetch_posts(user: User) -> Effect[Any, str, List[Post]]: ...

user_posts = fetch_user(123).flat_map(fetch_posts)

`map(f)` ¶

Transform the success value of this effect.

If this effect succeeds with value A, apply function f to get value B. If this effect fails, the error passes through unchanged.

Parameters:

Name	Type	Description	Default
`f`	`Callable[[A], B]`	Function to transform the success value	required

Returns:

Type	Description
`'Effect[R, E, B]'`	A new effect with transformed success type

Example

doubled = succeed(21).map(lambda x: x * 2)  # Effect[Any, None, int]
result = await doubled._run(Context())  # 42

`provide(layer)` ¶

Run this effect with additional services from a layer.

The layer is built, this effect runs with the enhanced context, then the layer is torn down - even if the effect fails.

Parameters:

Name	Type	Description	Default
`layer`	`_LayerLike`	Layer providing additional services	required

Returns:

Type	Description
`'Effect[Any, E, A]'`	Effect that no longer requires the layer's services

Example

effect_with_db = my_effect.provide(DatabaseLayer)
result = await effect_with_db._run(Context())  # DatabaseLayer services available

`Exit` `dataclass` ¶

Bases: Generic[E, A]

Represents the result of running an Effect.

An Exit can either be successful (with a value) or failed (with a cause). This is used internally by the Effect system to handle both success and failure cases.

Attributes:

Name	Type	Description
`success`	`bool`	True if the effect succeeded, False if it failed
`value`	`Optional[A]`	The success value if successful, None otherwise
`cause`	`Optional['Cause[E]']`	The failure cause if failed, None otherwise

`Failure` ¶

Bases: Exception, Generic[E]

Exception raised when an Effect fails.

This exception carries structured error information including the original error and any annotations that were added during effect composition.

Parameters:

Name	Type	Description	Default
`error`	`E`	The original error value	required
`annotations`	`Optional[list[str]]`	Optional list of annotation strings for debugging	`None`

`acquire_release(acquire, release, use)` ¶

Resource-safe acquire/release pattern (bracket operation).

Guarantees that the release function is called even if the use function fails or is interrupted. This is the foundation for safe resource management.

Parameters:

Name	Type	Description	Default
`acquire`	`Effect[Any, E, A]`	Effect to acquire the resource	required
`release`	`Callable[[A], Effect[Any, Any, Any]]`	Function to release the resource (called even on failure)	required
`use`	`Callable[[A], Effect[Any, E2, B]]`	Function to use the resource	required

Returns:

Type	Description
`Effect[Any, E \| E2, B]`	Effect that safely manages the resource lifecycle

Example

def with_file(path: str) -> Effect[Any, str, str]:
    return acquire_release(
        acquire=from_async(lambda: open(path, 'r')),
        release=lambda f: sync(f.close),
        use=lambda f: from_async(f.read)
    )

`annotate_cause(c, note)` ¶

Add an annotation to a Cause for debugging purposes.

Parameters:

Name	Type	Description	Default
`c`	`Cause[E]`	The cause to annotate	required
`note`	`str`	The annotation string to add	required

Returns:

Type	Description
`Cause[E]`	A new Cause with the added annotation

`attempt(thunk, on_error)` ¶

Safely execute a function that might throw exceptions.

Any exceptions are caught and converted to Effect failures using the on_error function.

Parameters:

Name	Type	Description	Default
`thunk`	`Callable[[], A]`	Function that might throw exceptions	required
`on_error`	`Callable[[BaseException], E]`	Function to convert exceptions to error values	required

Returns:

Type	Description
`Effect[Any, E, A]`	An effect that either succeeds or fails gracefully

Example

def risky_parse(text: str) -> int:
    return int(text)  # Might throw ValueError

safe_parse = attempt(
    lambda: risky_parse("not_a_number"),
    lambda ex: f"Parse error: {ex}"
)

`fail(e)` ¶

Create an effect that always fails with the given error.

Parameters:

Name	Type	Description	Default
`e`	`E`	The error to fail with	required

Returns:

Type	Description
`Effect[Any, E, Any]`	An effect that immediately fails with the error

Example

try:
    await fail("something went wrong")._run(Context())
except Failure as f:
    print(f.error)  # "something went wrong"

`for_each_par(items, f, parallelism=10)` ¶

Apply an effect-producing function to each item in parallel.

Processes items concurrently with limited parallelism. If any effect fails, all others are cancelled.

Parameters:

Name	Type	Description	Default
`items`	`Iterable[T]`	Collection of items to process	required
`f`	`Callable[[T], Effect[Any, E, A]]`	Function that converts each item to an effect	required
`parallelism`	`int`	Maximum number of concurrent operations (default: 10)	`10`

Returns:

Type	Description
`Effect[Any, E, List[A]]`	Effect that succeeds with list of all results in original order

Example

user_ids = [1, 2, 3, 4, 5]
users = await for_each_par(
    user_ids,
    fetch_user,
    parallelism=3  # Max 3 concurrent fetches
)._run(Context())

`from_async(thunk)` ¶

Convert an async function to an effect.

Parameters:

Name	Type	Description	Default
`thunk`	`Callable[[], Awaitable[A]]`	Async function that takes no arguments	required

Returns:

Type	Description
`Effect[Any, Any, A]`	An effect that runs the async function

Example

async def fetch_data():
    await asyncio.sleep(0.1)
    return "data"

effect = from_async(fetch_data)
result = await effect._run(Context())  # "data"

`race(e1, e2)` ¶

Run two effects concurrently, return the first to succeed.

Both effects start simultaneously. The first to succeed wins, and the other effect is cancelled.

Parameters:

Name	Type	Description	Default
`e1`	`Effect[Any, E, A]`	First effect to race	required
`e2`	`Effect[Any, E, A]`	Second effect to race	required

Returns:

Type	Description
`Effect[Any, E, A]`	Effect that succeeds with the result of whichever effect finishes first

Example

# Try primary service, but use backup if it's faster
result = await race(
    fetch_from_primary(),
    fetch_from_backup()
)._run(Context())

`succeed(a)` ¶

Create an effect that always succeeds with the given value.

Parameters:

Name	Type	Description	Default
`a`	`A`	The value to succeed with	required

Returns:

Type	Description
`Effect[Any, Any, A]`	An effect that immediately succeeds with the value

Example

result = await succeed(42)._run(Context())  # 42

`sync(thunk)` ¶

Convert a synchronous function to an effect.

Parameters:

Name	Type	Description	Default
`thunk`	`Callable[[], A]`	Synchronous function that takes no arguments	required

Returns:

Type	Description
`Effect[Any, Any, A]`	An effect that runs the function

Example

import random

random_effect = sync(lambda: random.randint(1, 100))
result = await random_effect._run(Context())  # Random number

`zip_par(e1, e2)` ¶

Run two effects concurrently and combine their results.

Both effects start simultaneously. If either fails, the other is cancelled. Returns a tuple of both results if both succeed.

Parameters:

Name	Type	Description	Default
`e1`	`Effect[Any, E, A]`	First effect to run	required
`e2`	`Effect[Any, E, B]`	Second effect to run	required

Returns:

Type	Description
`Effect[Any, E, Tuple[A, B]]`	Effect that succeeds with tuple of both results

Example

user_and_posts = await zip_par(
    fetch_user(123),
    fetch_posts(123)
)._run(Context())

user, posts = user_and_posts

Core API Reference¶

Core Types and Functions¶

effectpy.core ¶

Cause dataclass ¶

both(l, r) staticmethod ¶

die(ex) staticmethod ¶

fail(e) staticmethod ¶

interrupt() staticmethod ¶

render(indent='', include_traces=True) ¶

then(l, r) staticmethod ¶

Effect ¶

catch_all(f) ¶

flat_map(f) ¶

map(f) ¶

provide(layer) ¶

Exit dataclass ¶

Failure ¶

acquire_release(acquire, release, use) ¶

annotate_cause(c, note) ¶

attempt(thunk, on_error) ¶

fail(e) ¶

for_each_par(items, f, parallelism=10) ¶

from_async(thunk) ¶

race(e1, e2) ¶

succeed(a) ¶

sync(thunk) ¶

zip_par(e1, e2) ¶

`effectpy.core` ¶

`Cause` `dataclass` ¶

`both(l, r)` `staticmethod` ¶

`die(ex)` `staticmethod` ¶

`fail(e)` `staticmethod` ¶

`interrupt()` `staticmethod` ¶

`render(indent='', include_traces=True)` ¶

`then(l, r)` `staticmethod` ¶

`Effect` ¶

`catch_all(f)` ¶

`flat_map(f)` ¶

`map(f)` ¶

`provide(layer)` ¶

`Exit` `dataclass` ¶

`Failure` ¶

`acquire_release(acquire, release, use)` ¶

`annotate_cause(c, note)` ¶

`attempt(thunk, on_error)` ¶

`fail(e)` ¶

`for_each_par(items, f, parallelism=10)` ¶

`from_async(thunk)` ¶

`race(e1, e2)` ¶

`succeed(a)` ¶

`sync(thunk)` ¶

`zip_par(e1, e2)` ¶