AsyncResult Monad

The AsyncResult monad extends Result with async operations, supporting awaitable computations with error handling. It combines the benefits of async/await with explicit error handling.

AsyncResult[T, E] wraps a Result[T, E] and provides async operations. All operations that might be async are awaitable, and it converts to/from regular Result.

Creating AsyncResults

Creating Success Values

The AsyncResult.ok method creates an AsyncResult containing a success value.

from better_py import AsyncResult

success = AsyncResult.ok(42)

Use this when you have a successful result to wrap.

Creating Error Values

The AsyncResult.error method creates an AsyncResult containing an error.

from better_py import AsyncResult

failure = AsyncResult.error("Something went wrong")

Use this when you need to represent a failed computation.

Creating from Values

The AsyncResult.from_value method creates an AsyncResult from a value, treating None as an error.

from better_py import AsyncResult

AsyncResult.from_value(42)              # Ok(42)
AsyncResult.from_value(None, "error")  # Error('error')

Use this to convert existing values into AsyncResult with custom error handling.

Checking Variants

Checking for Success

The is_ok_async method checks if the AsyncResult contains a success value.

from better_py import AsyncResult

await AsyncResult.ok(42).is_ok_async()      # True
await AsyncResult.error("bad").is_ok_async()  # False

Use this to check for successful computations.

Checking for Errors

The is_error_async method checks if the AsyncResult contains an error.

from better_py import AsyncResult

await AsyncResult.ok(42).is_error_async()     # False
await AsyncResult.error("bad").is_error_async()  # True

Use this to check for failed computations.

Extracting Values

Getting Values or Errors

The unwrap method returns the success value or raises an error if failed.

from better_py import AsyncResult

await AsyncResult.ok(42).unwrap()          # 42
await AsyncResult.error("bad").unwrap()    # Raises ValueError

Use this only when you're certain the AsyncResult contains a success value.

Getting Values or Defaults

The unwrap_or_else method returns the success value or computes a default.

from better_py import AsyncResult

await AsyncResult.ok(42).unwrap_or_else(lambda: 0)      # 42
await AsyncResult.error("bad").unwrap_or_else(lambda: 0)  # 0

Use this when you need safe access to the value with a fallback.

Getting Error Values

The unwrap_error method returns the error value or raises an error if successful.

from better_py import AsyncResult

await AsyncResult.error("bad").unwrap_error()  # "bad"
await AsyncResult.ok(42).unwrap_error()        # Raises ValueError

Use this when you need to access the error value for logging or handling.

Transforming Values

Mapping Success Values

The map method transforms the success value, preserving errors.

from better_py import AsyncResult

AsyncResult.ok(5).map(lambda x: x * 2)           # Ok(10)
AsyncResult.error("bad").map(lambda x: x * 2)  # Error("bad")

When map is called on Error, it returns the Error unchanged.

Mapping Async Functions

The map_async method applies an async function to the success value.

from better_py import AsyncResult

async def fetch(x: int) -> str:
    return await api.get(f"/items/{x}")

await AsyncResult.ok(5).map_async(fetch)      # Ok(result)
await AsyncResult.error("bad").map_async(fetch)  # Error("bad")

Use this when the transformation function is async.

Transforming Errors

The map_error method transforms the error value, preserving successes.

from better_py import AsyncResult

AsyncResult.error("bad").map_error(str.upper)  # Error("BAD")
AsyncResult.ok(42).map_error(str.upper)       # Ok(42)

Use this to format, enrich, or normalize errors.

Chaining Operations

Binding Async Operations

The bind method chains AsyncResult-returning operations, short-circuiting on errors.

from better_py import AsyncResult

async def fetch_user(user_id: int) -> AsyncResult[dict, str]:
    user = await database.fetch(user_id)
    if user:
        return AsyncResult.ok(user)
    return AsyncResult.error("User not found")

async def get_orders(user: dict) -> AsyncResult[list, str]:
    orders = await database.fetch_orders(user["id"])
    return AsyncResult.ok(orders)

# Chain operations
orders = await fetch_user(1).bind(get_orders)  # Ok([...]) or Error(...)

Use bind for sequential operations where each step depends on the previous one.

Recovering from Errors

The recover method transforms errors into successes using a recovery function.

from better_py import AsyncResult

await AsyncResult.error("bad").recover(lambda e: 0)  # Ok(0)
await AsyncResult.ok(42).recover(lambda e: 0)       # Ok(42)

Use recover to provide default values or fallback logic when operations fail.

Converting to Result

Sync Conversion

The to_result method converts AsyncResult to a regular Result.

from better_py import AsyncResult, Ok, Error

AsyncResult.ok(42).to_result()            # Ok(42)
AsyncResult.error("bad").to_result()     # Error("bad")

Use this when you need to integrate with non-async code that uses Result.

Real-World Pattern: Async API Pipeline

from better_py import AsyncResult

async def fetch_user(user_id: int) -> AsyncResult[dict, str]:
    try:
        user = await database.query(f"SELECT * FROM users WHERE id = {user_id}")
        if user:
            return AsyncResult.ok(user)
        return AsyncResult.error("User not found")
    except DatabaseError as e:
        return AsyncResult.error(str(e))

async def validate_user(user: dict) -> AsyncResult[dict, str]:
    if not user.get("email"):
        return AsyncResult.error("Email required")
    return AsyncResult.ok(user)

async def send_notification(user: dict) -> AsyncResult[bool, str]:
    try:
        await notify.send(user["email"])
        return AsyncResult.ok(True)
    except NotificationError as e:
        return AsyncResult.error(str(e))

async def onboard_user(user_id: int) -> AsyncResult[dict, str]:
    return (await fetch_user(user_id)
        .bind(validate_user)
        .bind(send_notification)
        .map(lambda _: {"status": "onboarded"}))

result = await onboard_user(1)
# Ok({"status": "onboarded"}) or Error("...")

This pattern shows AsyncResult's power: async operations that can fail are handled gracefully, errors are propagated through the chain, and the pipeline is readable without nested try/except blocks.

When to Use

Use AsyncResult when:

Working with async code that can fail
You need awaitable error handling
You're using asyncio
You want to chain async operations with errors
You need explicit error types instead of exceptions

Don't use AsyncResult when:

You're not using async (use Result instead)
Operations always succeed (use plain types)
You only need optional values (use AsyncMaybe instead)
You're doing simple async operations (plain async/await is fine)