tessl/pypi-returns

Functional programming library providing type-safe containers for error handling, side effects, and composable operations with monadic patterns.

—

Pending

Overview

Eval results

Files

Conversions

Name: tessl/pypi-returns
Author: tessl

Utilities for converting between different container types and integrating with external libraries and legacy code. These functions enable seamless interoperability between monadic containers and traditional Python code.

Capabilities

Container Type Conversions

Functions for converting between different container types while preserving semantic meaning.

def result_to_maybe(result_container: Result[T, E]) -> Maybe[T]:
    """Convert Result to Maybe, discarding error information"""

def maybe_to_result(maybe_container: Maybe[T], default_error: E = None) -> Result[T, E]:
    """Convert Maybe to Result, providing error for Nothing case"""

def io_to_result(io_container: IO[T]) -> IOResult[T, Exception]:
    """Convert IO to IOResult by catching exceptions"""

def result_to_io(result_container: Result[T, E]) -> IO[Result[T, E]]:
    """Lift Result into IO context"""

def future_to_future_result(future_container: Future[T]) -> FutureResult[T, Never]:
    """Convert Future to FutureResult (never fails)"""

def future_result_to_future(future_result: FutureResult[T, E]) -> Future[Result[T, E]]:
    """Convert FutureResult to Future of Result"""

Usage examples:

from returns.converters import result_to_maybe, maybe_to_result, io_to_result
from returns.result import Success, Failure
from returns.maybe import Some, Nothing
from returns.io import IO

# Result to Maybe conversion
success_result = Success(42)
maybe_from_success = result_to_maybe(success_result)  # Some(42)

failure_result = Failure("error occurred")
maybe_from_failure = result_to_maybe(failure_result)  # Nothing

# Maybe to Result conversion
some_value = Some(42)
result_from_some = maybe_to_result(some_value, "no value provided")  # Success(42)

nothing_value = Nothing
result_from_nothing = maybe_to_result(nothing_value, "no value provided")  # Failure("no value provided")

# Using default error
result_with_default = maybe_to_result(nothing_value)  # Failure(None)

# IO to Result conversion (exception handling)
def risky_io_operation() -> str:
    raise ValueError("Something went wrong")

io_operation = IO(risky_io_operation)
io_result = io_to_result(io_operation)
result = io_result()  # Failure(ValueError("Something went wrong"))

Flattening Operations

Utilities for flattening nested container structures.

def flatten(container: Container[Container[T]]) -> Container[T]:
    """Join two nested containers together"""

def flatten_result(nested: Result[Result[T, E], E]) -> Result[T, E]:
    """Flatten nested Result containers"""

def flatten_maybe(nested: Maybe[Maybe[T]]) -> Maybe[T]:
    """Flatten nested Maybe containers"""

def flatten_io(nested: IO[IO[T]]) -> IO[T]:
    """Flatten nested IO containers"""

def flatten_future(nested: Future[Future[T]]) -> Future[T]:
    """Flatten nested Future containers"""

Usage examples:

from returns.converters import flatten, flatten_result, flatten_maybe
from returns.result import Success, Failure
from returns.maybe import Some, Nothing

# Generic flatten
nested_success = Success(Success(42))
flattened = flatten(nested_success)  # Success(42)

nested_failure_outer = Failure("outer error")
flattened_outer = flatten(nested_failure_outer)  # Failure("outer error")

nested_failure_inner = Success(Failure("inner error"))
flattened_inner = flatten(nested_failure_inner)  # Failure("inner error")

# Specific flatten operations
nested_maybe = Some(Some("hello"))
flattened_maybe = flatten_maybe(nested_maybe)  # Some("hello")

nested_nothing = Some(Nothing)
flattened_nothing = flatten_maybe(nested_nothing)  # Nothing

# Practical example: nested operations
def validate_input(s: str) -> Result[str, str]:
    return Success(s.strip()) if s.strip() else Failure("Empty input")

def parse_number(s: str) -> Result[Result[int, str], str]:
    validated = validate_input(s)
    if isinstance(validated, Success):
        try:
            return Success(Success(int(validated.unwrap())))
        except ValueError:
            return Success(Failure("Not a number"))
    return Failure("Validation failed")

# Without flatten: Result[Result[int, str], str]
nested_result = parse_number("42")

# With flatten: Result[int, str]  
flattened_result = flatten(nested_result)  # Success(42)

Legacy Code Integration

Utilities for integrating with traditional Python code that doesn't use monadic containers.

def unsafe_perform_io(io_container: IO[T]) -> T:
    """Execute IO operation and return raw value (unsafe)"""

def unwrap_or_failure(container: Container[T, E], failure_value: E) -> T | E:
    """Unwrap container value or return failure value"""

def to_optional(container: Maybe[T]) -> T | None:
    """Convert Maybe to optional value (T | None)"""

def from_optional(value: T | None) -> Maybe[T]:
    """Convert optional value to Maybe"""

def to_exception(container: Result[T, E]) -> T:
    """Convert Result to value or raise exception"""

def from_exception(func: Callable[..., T]) -> Callable[..., Result[T, Exception]]:
    """Convert exception-throwing function to Result-returning function"""

Usage examples:

from returns.converters import (
    unsafe_perform_io, to_optional, from_optional, 
    to_exception, from_exception
)
from returns.io import IO
from returns.maybe import Some, Nothing
from returns.result import Success, Failure, safe

# Unsafe IO execution (for integration with legacy code)
def read_config() -> str:
    return "config_data"

io_config = IO(read_config)
config_data = unsafe_perform_io(io_config)  # "config_data"

# Maybe to optional conversion
some_value = Some(42)
optional_value = to_optional(some_value)  # 42

nothing_value = Nothing  
optional_none = to_optional(nothing_value)  # None

# Optional to Maybe conversion
optional_data = 42
maybe_data = from_optional(optional_data)  # Some(42)

none_data = None
maybe_none = from_optional(none_data)  # Nothing

# Result to exception conversion (for legacy APIs)
success_result = Success(42)
value = to_exception(success_result)  # 42

failure_result = Failure(ValueError("Something went wrong"))
# value = to_exception(failure_result)  # Raises ValueError("Something went wrong")

# Exception to Result conversion
def legacy_function(x: int) -> int:
    if x < 0:
        raise ValueError("Negative value")
    return x * 2

safe_function = from_exception(legacy_function)
result1 = safe_function(5)   # Success(10)
result2 = safe_function(-5)  # Failure(ValueError("Negative value"))

Async Conversions

Utilities for converting between sync and async containers.

def sync_to_async(container: Container[T]) -> Future[Container[T]]:
    """Convert sync container to async Future"""

async def async_to_sync(future_container: Future[T]) -> T:
    """Convert async Future to sync value (requires await)"""

def result_to_future_result(result: Result[T, E]) -> FutureResult[T, E]:
    """Convert Result to FutureResult"""

async def future_result_to_result(future_result: FutureResult[T, E]) -> Result[T, E]:
    """Convert FutureResult to Result (requires await)"""

def io_to_future(io_container: IO[T]) -> Future[T]:
    """Convert IO to Future"""

def future_to_io(future_container: Future[T]) -> IO[Future[T]]:
    """Convert Future to IO of Future"""

Usage examples:

import asyncio
from returns.converters import (
    sync_to_async, result_to_future_result, 
    future_result_to_result, io_to_future
)
from returns.result import Success, Failure
from returns.future import Future, FutureResult
from returns.io import IO

# Sync to async conversion
sync_result = Success(42)
async_result = sync_to_async(sync_result)
future_success = await async_result  # Success(42)

# Result to FutureResult
result = Success(42)
future_result = result_to_future_result(result)
converted_result = await future_result  # Success(42)

# FutureResult to Result
async def async_operation() -> Result[int, str]:
    return Success(42)

future_result = FutureResult(async_operation())
sync_result = await future_result_to_result(future_result)  # Success(42)

# IO to Future
def io_operation() -> str:
    return "data from IO"

io_container = IO(io_operation)
future_container = io_to_future(io_container)
result = await future_container  # "data from IO"

Type Narrowing

Utilities for type narrowing and runtime type checking with containers.

def is_success(container: Result[T, E]) -> bool:
    """Check if Result is Success"""

def is_failure(container: Result[T, E]) -> bool:
    """Check if Result is Failure"""

def is_some(container: Maybe[T]) -> bool:
    """Check if Maybe is Some"""

def is_nothing(container: Maybe[T]) -> bool:
    """Check if Maybe is Nothing"""

def success_value(container: Result[T, E]) -> T | None:
    """Extract value from Success, None if Failure"""

def failure_value(container: Result[T, E]) -> E | None:
    """Extract error from Failure, None if Success"""

def some_value(container: Maybe[T]) -> T | None:
    """Extract value from Some, None if Nothing"""

Usage examples:

from returns.converters import (
    is_success, is_failure, success_value, failure_value,
    is_some, some_value
)
from returns.result import Success, Failure
from returns.maybe import Some, Nothing

# Result type checking
success = Success(42)
failure = Failure("error")

print(is_success(success))  # True
print(is_failure(success))  # False
print(is_success(failure))  # False
print(is_failure(failure))  # True

# Value extraction
success_val = success_value(success)    # 42
success_val_from_failure = success_value(failure)  # None
failure_val = failure_value(failure)    # "error"
failure_val_from_success = failure_value(success)  # None

# Maybe type checking
some = Some(42)
nothing = Nothing

print(is_some(some))     # True
print(is_nothing(some))  # False
print(is_some(nothing))  # False
print(is_nothing(nothing))  # True

# Maybe value extraction
some_val = some_value(some)     # 42
some_val_from_nothing = some_value(nothing)  # None

Integration Patterns

Gradual Migration

from returns.converters import from_exception, to_exception
from returns.result import Result

# Wrap legacy function
def legacy_divide(a: int, b: int) -> float:
    return a / b  # May raise ZeroDivisionError

# Create safe wrapper
safe_divide = from_exception(legacy_divide)

# Use in functional pipeline
def calculate_average(numbers: list[int]) -> Result[float, Exception]:
    if not numbers:
        return Failure(ValueError("Empty list"))
    
    total = sum(numbers)
    count = len(numbers)
    
    return safe_divide(total, count)

# Convert back for legacy code when needed
def legacy_api_endpoint(numbers: list[int]) -> float:
    result = calculate_average(numbers)
    return to_exception(result)  # Raises exception if failed

External Library Integration

from returns.converters import from_optional, to_optional
from returns.maybe import Maybe
import json
from typing import Any

# Integrate with libraries that return None for missing values
def parse_json_field(data: str, field: str) -> Maybe[Any]:
    try:
        parsed = json.loads(data)
        field_value = parsed.get(field)  # Returns None if missing
        return from_optional(field_value)
    except json.JSONDecodeError:
        return Nothing

# Use in functional pipeline
def process_config(json_str: str) -> Maybe[str]:
    return (
        parse_json_field(json_str, "database")
        .bind(lambda db_config: parse_json_field(str(db_config), "url"))
        .map(str)
    )

# Convert back to optional for external API
def get_database_url(json_str: str) -> str | None:
    result = process_config(json_str)
    return to_optional(result)

Conversion utilities provide essential bridges between the functional programming paradigm of Returns and traditional Python code, enabling gradual adoption and seamless integration with existing codebases and libraries.

Install with Tessl CLI