CtrlK
BlogDocsLog inGet started
Tessl Logo

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

core-containers.mddocs/

Core Containers

Essential monadic containers that form the foundation of functional programming patterns in Returns. These containers provide type-safe ways to handle errors, optional values, side effects, and asynchronous operations.

Capabilities

Result Container

The Result container represents computations that can either succeed with a value or fail with an error. It enables railway-oriented programming where operations are chained without explicit error handling at each step.

class Result[T, E]:
    """Abstract base class for Success and Failure"""
    
class Success[T]:
    """Represents successful computation"""
    def __init__(self, value: T): ...
    def bind(self, func: Callable[[T], Result[U, E]]) -> Result[U, E]: ...
    def map(self, func: Callable[[T], U]) -> Result[U, E]: ...
    def apply(self, wrapped_func: Result[Callable[[T], U], E]) -> Result[U, E]: ...
    def alt(self, func: Callable[[E], Result[T, F]]) -> Result[T, F]: ...
    def swap(self) -> Result[E, T]: ...
    def unwrap(self) -> T: ...
    def value_or(self, default: U) -> T | U: ...
    def lash(self, func: Callable[[E], Result[T, F]]) -> Result[T, F]: ...
    
class Failure[E]:
    """Represents failed computation"""
    def __init__(self, error: E): ...
    def bind(self, func: Callable[[T], Result[U, E]]) -> Result[U, E]: ...
    def map(self, func: Callable[[T], U]) -> Result[U, E]: ...
    def apply(self, wrapped_func: Result[Callable[[T], U], E]) -> Result[U, E]: ...
    def alt(self, func: Callable[[E], Result[T, F]]) -> Result[T, F]: ...
    def swap(self) -> Result[E, T]: ...
    def failure(self) -> E: ...
    def value_or(self, default: U) -> U: ...
    def lash(self, func: Callable[[E], Result[T, F]]) -> Result[T, F]: ...

# Type alias for Result with Exception
ResultE[T] = Result[T, Exception]

# Decorator functions
def safe(func: Callable[..., T]) -> Callable[..., Result[T, Exception]]:
    """Convert exception-throwing function to Result"""

def attempt(func: Callable[..., T]) -> Callable[..., Result[T, Exception]]:
    """Similar to safe but wraps argument on exception"""

# Additional methods and aliases
bind_result = bind  # Alias for bind method, part of ResultBased interface

# Class methods for construction
@classmethod
def from_value(cls, value: T) -> Result[T, E]:
    """Create Success from value"""
    
@classmethod  
def from_failure(cls, error: E) -> Result[T, E]:
    """Create Failure from error"""
    
@classmethod
def do(cls, expr: Generator[T, None, None]) -> Result[T, E]:
    """Support for do-notation (generator comprehensions)"""

# Iterator support for do-notation
def __iter__(self) -> Iterator[T]:
    """Enable do-notation syntax"""

# Pattern matching support  
__match_args__ = ('_inner_value',)

# Type-safe equality
equals = container_equality

Usage examples:

from returns.result import Success, Failure, safe, Result

# Creating Result instances
success_result = Success(42)
failure_result = Failure("Something went wrong")

# Using safe decorator
@safe
def divide(a: int, b: int) -> float:
    return a / b

result = divide(10, 2)    # Success(5.0)
result = divide(10, 0)    # Failure(ZeroDivisionError(...))

# Chaining operations
def double(x: int) -> Result[int, str]:
    return Success(x * 2)

def validate_positive(x: int) -> Result[int, str]:
    return Success(x) if x > 0 else Failure("Must be positive")

result = Success(5).bind(double).bind(validate_positive)  # Success(10)

Maybe Container

The Maybe container handles optional values in a type-safe way, eliminating null pointer exceptions and making the absence of values explicit in the type system.

class Maybe[T]:
    """Abstract base class for Some and Nothing"""
    
class Some[T]:
    """Represents presence of value"""
    def __init__(self, value: T): ...
    def bind(self, func: Callable[[T], Maybe[U]]) -> Maybe[U]: ...
    def map(self, func: Callable[[T], U]) -> Maybe[U]: ...
    def apply(self, wrapped_func: Maybe[Callable[[T], U]]) -> Maybe[U]: ...
    def unwrap(self) -> T: ...
    def value_or(self, default: U) -> T | U: ...
    def or_else_call(self, func: Callable[[], U]) -> T | U: ...
    def bind_optional(self, func: Callable[[T], U | None]) -> Maybe[U]: ...
    def lash(self, func: Callable[[None], Maybe[T]]) -> Maybe[T]: ...

class _Nothing:
    """Represents absence of value (singleton)"""
    def bind(self, func: Callable[[T], Maybe[U]]) -> Maybe[U]: ...
    def map(self, func: Callable[[T], U]) -> Maybe[U]: ...
    def apply(self, wrapped_func: Maybe[Callable[[T], U]]) -> Maybe[U]: ...
    def value_or(self, default: U) -> U: ...
    def or_else_call(self, func: Callable[[], U]) -> U: ...
    def bind_optional(self, func: Callable[[T], U | None]) -> Maybe[U]: ...
    def lash(self, func: Callable[[None], Maybe[T]]) -> Maybe[T]: ...
    def failure(self) -> None: ...

# Singleton instance
Nothing: _Nothing

# Decorator function
def maybe(func: Callable[..., T | None]) -> Callable[..., Maybe[T]]:
    """Convert None-returning function to Maybe"""

# Class methods for construction
@classmethod
def from_value(cls, value: T) -> Maybe[T]:
    """Create Some from value"""
    
@classmethod
def from_optional(cls, value: T | None) -> Maybe[T]:
    """Create Some or Nothing from optional value"""
    
@classmethod
def do(cls, expr: Generator[T, None, None]) -> Maybe[T]:
    """Support for do-notation (generator comprehensions)"""

# Iterator support and additional methods
def __iter__(self) -> Iterator[T]:
    """Enable do-notation syntax"""
    
def __bool__(self) -> bool:
    """Boolean conversion (Some is True, Nothing is False)"""

# Type-safe equality
equals = container_equality

Usage examples:

from returns.maybe import Some, Nothing, maybe, Maybe

# Creating Maybe instances
some_value = Some(42)
no_value = Nothing

# Using maybe decorator
@maybe
def find_user(user_id: int) -> dict | None:
    return {"id": user_id, "name": "John"} if user_id > 0 else None

user = find_user(1)   # Some({"id": 1, "name": "John"})
user = find_user(-1)  # Nothing

# Chaining operations
def get_name(user: dict) -> Maybe[str]:
    return Some(user.get("name")) if user.get("name") else Nothing

name = find_user(1).bind(get_name)  # Some("John")

IO Container

The IO container represents impure operations (side effects) while maintaining referential transparency. It defers execution until explicitly run.

class IO[T]:
    """Container for impure operations that never fail"""
    def __init__(self, value: T): ...
    def bind(self, func: Callable[[T], IO[U]]) -> IO[U]: ...
    def map(self, func: Callable[[T], U]) -> IO[U]: ...
    def apply(self, wrapped_func: IO[Callable[[T], U]]) -> IO[U]: ...
    
    # Factory methods
    @classmethod
    def from_value(cls, value: T) -> IO[T]: ...
    @classmethod
    def from_io(cls, value: IO[T]) -> IO[T]: ...
    
    # Aliases and additional methods
    bind_io = bind  # Alias for bind method
    
    # Do-notation support
    @classmethod
    def do(cls, expr: Generator[T, None, None]) -> IO[T]: ...
    def __iter__(self) -> Iterator[T]: ...

class IOResult[T, E]:
    """Container for impure operations that can fail"""
    def __init__(self, value: Result[T, E]): ...
    def bind(self, func: Callable[[T], IOResult[U, E]]) -> IOResult[U, E]: ...
    def map(self, func: Callable[[T], U]) -> IOResult[U, E]: ...
    def apply(self, wrapped_func: IOResult[Callable[[T], U], E]) -> IOResult[U, E]: ...
    def alt(self, func: Callable[[E], IOResult[T, F]]) -> IOResult[T, F]: ...
    def lash(self, func: Callable[[E], IOResult[T, F]]) -> IOResult[T, F]: ...
    def swap(self) -> IOResult[E, T]: ...
    def unwrap(self) -> T: ...
    def value_or(self, default: U) -> T | U: ...
    def failure(self) -> E: ...
    
    # Factory methods
    @classmethod
    def from_result(cls, value: Result[T, E]) -> IOResult[T, E]: ...
    @classmethod
    def from_io(cls, value: IO[T]) -> IOResult[T, E]: ...
    @classmethod
    def from_value(cls, value: T) -> IOResult[T, E]: ...
    @classmethod
    def from_failure(cls, error: E) -> IOResult[T, E]: ...
    
    # Specialized bind methods
    def bind_result(self, func: Callable[[T], Result[U, E]]) -> IOResult[U, E]: ...
    def bind_io(self, func: Callable[[T], IO[U]]) -> IOResult[U, E]: ...
    def compose_result(self, func: Callable[[T], IOResult[U, E]]) -> IOResult[U, E]: ...

class IOSuccess[T]:
    """Success variant of IOResult"""
    def __init__(self, value: T): ...

class IOFailure[E]:
    """Failure variant of IOResult"""  
    def __init__(self, error: E): ...

# Type alias
IOResultE[T] = IOResult[T, Exception]

# Decorator functions
def impure(func: Callable[..., T]) -> Callable[..., IO[T]]:
    """Mark sync function as returning IO"""

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

Usage examples:

from returns.io import IO, IOResult, impure, impure_safe

# Creating IO instances
def read_file() -> str:
    with open("data.txt") as f:
        return f.read()

io_operation = IO(read_file)
content = io_operation()  # Execute the operation

# Using decorators
@impure
def print_message(msg: str) -> None:
    print(msg)

@impure_safe  
def divide_io(a: int, b: int) -> float:
    return a / b

io_print = print_message("Hello")  # IO[None]
io_result = divide_io(10, 2)       # IOResult[float, Exception]

Future Container

The Future container represents asynchronous operations, providing composable async programming with type safety.

class Future[T]:
    """Container for async operations that never fail"""
    def __init__(self, awaitable: Awaitable[T]): ...
    def bind(self, func: Callable[[T], Future[U]]) -> Future[U]: ...
    def map(self, func: Callable[[T], U]) -> Future[U]: ...
    def apply(self, wrapped_func: Future[Callable[[T], U]]) -> Future[U]: ...
    async def awaitable(self) -> T: ...
    
    # Additional bind methods
    def bind_async(self, func: Callable[[T], Awaitable[U]]) -> Future[U]: ...
    def bind_awaitable(self, func: Callable[[T], Awaitable[U]]) -> Future[U]: ...
    def bind_io(self, func: Callable[[T], IO[U]]) -> Future[U]: ...
    
    # Factory methods
    @classmethod
    def from_value(cls, value: T) -> Future[T]: ...
    @classmethod
    def from_io(cls, value: IO[T]) -> Future[T]: ...
    @classmethod
    def from_future_result(cls, value: FutureResult[T, E]) -> Future[T]: ...
    
    # Do-notation support (async)
    @classmethod
    def do(cls, expr: AsyncGenerator[T, None]) -> Future[T]: ...
    def __aiter__(self): ...

class FutureResult[T, E]:
    """Container for async operations that can fail"""
    def __init__(self, awaitable: Awaitable[Result[T, E]]): ...
    def bind(self, func: Callable[[T], FutureResult[U, E]]) -> FutureResult[U, E]: ...
    def map(self, func: Callable[[T], U]) -> FutureResult[U, E]: ...
    def apply(self, wrapped_func: FutureResult[Callable[[T], U], E]) -> FutureResult[U, E]: ...
    def alt(self, func: Callable[[E], FutureResult[T, F]]) -> FutureResult[T, F]: ...
    def lash(self, func: Callable[[E], FutureResult[T, F]]) -> FutureResult[T, F]: ...
    def swap(self) -> FutureResult[E, T]: ...
    async def awaitable(self) -> Result[T, E]: ...
    
    # Additional bind methods  
    def bind_async(self, func: Callable[[T], Awaitable[U]]) -> FutureResult[U, E]: ...
    def bind_awaitable(self, func: Callable[[T], Awaitable[U]]) -> FutureResult[U, E]: ...
    def bind_result(self, func: Callable[[T], Result[U, E]]) -> FutureResult[U, E]: ...
    def bind_ioresult(self, func: Callable[[T], IOResult[U, E]]) -> FutureResult[U, E]: ...
    def bind_io(self, func: Callable[[T], IO[U]]) -> FutureResult[U, E]: ...
    def bind_future(self, func: Callable[[T], Future[U]]) -> FutureResult[U, E]: ...
    def bind_async_future(self, func: Callable[[T], Awaitable[Future[U]]]) -> FutureResult[U, E]: ...
    
    # Advanced composition
    def compose_result(self, func: Callable[[T], FutureResult[U, E]]) -> FutureResult[U, E]: ...
    
    # Factory methods
    @classmethod
    def from_result(cls, value: Result[T, E]) -> FutureResult[T, E]: ...
    @classmethod
    def from_future(cls, value: Future[T]) -> FutureResult[T, E]: ...
    @classmethod
    def from_io(cls, value: IO[T]) -> FutureResult[T, E]: ...
    @classmethod
    def from_ioresult(cls, value: IOResult[T, E]) -> FutureResult[T, E]: ...

# Type alias
FutureResultE[T] = FutureResult[T, Exception]

# Constructor functions
def FutureSuccess(value: T) -> FutureResult[T, E]:
    """Create successful FutureResult"""

def FutureFailure(error: E) -> FutureResult[T, E]:
    """Create failed FutureResult"""

# Decorator functions
def future(func: Callable[..., Awaitable[T]]) -> Callable[..., Future[T]]:
    """Turn coroutine into Future"""

def future_safe(func: Callable[..., Awaitable[T]]) -> Callable[..., FutureResult[T, Exception]]:
    """Convert exception-throwing coroutine to FutureResult"""

def asyncify(func: Callable[..., T]) -> Callable[..., Awaitable[T]]:
    """Turn sync function into async one"""

async def async_identity(instance: T) -> T:
    """Async version of identity function"""

Usage examples:

import asyncio
from returns.future import Future, FutureResult, future, future_safe

# Creating Future instances
async def fetch_data() -> str:
    await asyncio.sleep(1)
    return "data"

future_op = Future(fetch_data())
result = await future_op.awaitable()  # "data"

# Using decorators
@future
async def async_operation(x: int) -> int:
    await asyncio.sleep(0.1)
    return x * 2

@future_safe
async def risky_operation(x: int) -> int:  
    if x < 0:
        raise ValueError("Negative value")
    return x * 2

# Chaining async operations
async def process():
    result = await (
        async_operation(5)
        .bind(lambda x: async_operation(x + 1))
    ).awaitable()
    return result  # 12

Common Methods

All containers implement these core methods:

  • bind(func): Monadic bind operation for chaining
  • map(func): Transform successful/present values
  • apply(wrapped_func): Apply wrapped functions
  • iter(): Enable for-comprehension syntax
  • from_success(value) (classmethod): Create successful container
  • from_failure(error) (classmethod): Create failed container (where applicable)

Type Safety

All containers are fully typed with generic parameters and support mypy type checking. The library provides extensive type definitions and follows PEP 561 standards for type information distribution.

Install with Tessl CLI

npx tessl i tessl/pypi-returns

docs

async-operations.md

container-methods.md

context-operations.md

conversions.md

core-containers.md

development-tools.md

functional-utilities.md

index.md

iteration-utilities.md

pointfree.md

trampolines.md

unsafe-operations.md

tile.json