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

Point-free Operations

Name: tessl/pypi-returns
Author: tessl

Point-free style functions that enable composition without explicitly naming intermediate values, supporting functional programming idioms and container manipulation through higher-order functions.

Capabilities

Core Point-free Functions

Essential point-free operations that work with all container types for monadic composition.

def bind(func: Callable[[T], Container[U]]) -> Callable[[Container[T]], Container[U]]:
    """Monadic bind operation (flatMap)"""

def map_(func: Callable[[T], U]) -> Callable[[Container[T]], Container[U]]:  
    """Transform successful/present values"""

def apply(wrapped_func: Container[Callable[[T], U]]) -> Callable[[Container[T]], Container[U]]:
    """Apply wrapped function to wrapped value"""

def alt(func: Callable[[E], F]) -> Callable[[Container[T, E]], Container[T, F]]:
    """Transform error values"""

def lash(func: Callable[[E], Container[T, F]]) -> Callable[[Container[T, E]], Container[T, F]]:
    """Handle error cases with recovery"""

def swap() -> Callable[[Container[T, E]], Container[E, T]]:
    """Swap success and failure values"""

Usage examples:

from returns.pointfree import bind, map_, alt, lash
from returns.result import Success, Failure, Result
from returns.pipeline import flow

# Point-free bind
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")

# Traditional style
result = Success(5).bind(double).bind(validate_positive)

# Point-free style  
pipeline = flow(
    Success(5),
    bind(double),
    bind(validate_positive)
)

# Point-free map
def add_one(x: int) -> int:
    return x + 1

transform = map_(add_one)
result = transform(Success(42))  # Success(43)

# Point-free alt for error transformation
def format_error(error: str) -> str:
    return f"Error: {error}"

handle_error = alt(format_error)
result = handle_error(Failure("invalid input"))  # Failure("Error: invalid input")

Specialized Bind Operations

Type-specific bind operations for different container combinations.

def bind_result(func: Callable[[T], Result[U, E]]) -> Callable[[Container[T]], Container[U, E]]:
    """Bind operation specialized for Result"""

def bind_io(func: Callable[[T], IO[U]]) -> Callable[[Container[T]], Container[U]]:
    """Bind operation specialized for IO"""

def bind_future(func: Callable[[T], Future[U]]) -> Callable[[Container[T]], Container[U]]:
    """Bind operation specialized for Future"""

def bind_async(func: Callable[[T], Awaitable[U]]) -> Callable[[Container[T]], Awaitable[Container[U]]]:
    """Bind operation for async functions"""

def bind_context(func: Callable[[T], RequiresContext[U, Deps]]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[U, Deps]]:
    """Bind operation for context-dependent computations"""

Usage examples:

from returns.pointfree import bind_result, bind_io, bind_context
from returns.result import Success, safe
from returns.io import IO
from returns.context import RequiresContext

# Specialized Result bind
@safe
def parse_int(s: str) -> int:
    return int(s)

@safe  
def divide_by_two(x: int) -> float:
    return x / 2

process_string = flow(
    Success("42"),
    bind_result(parse_int),
    bind_result(divide_by_two)
)  # Success(21.0)

# IO bind
def read_file(path: str) -> IO[str]:
    return IO(lambda: open(path).read())

def process_content(content: str) -> IO[int]:
    return IO(lambda: len(content))

process_file = flow(
    IO(lambda: "data.txt"),
    bind_io(read_file),
    bind_io(process_content)
)

# Context bind
def get_config() -> RequiresContext[dict, dict]:
    return RequiresContext(lambda env: env["config"])

def get_database_url(config: dict) -> RequiresContext[str, dict]:
    return RequiresContext(lambda env: config["db_url"])

get_db_url = flow(
    RequiresContext.ask(),
    bind_context(get_config),
    bind_context(get_database_url)
)

Context Manipulation

Functions for working with context-dependent computations.

def modify_env(func: Callable[[Deps], NewDeps]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[T, NewDeps]]:
    """Modify the environment/context"""

def local(func: Callable[[Deps], NewDeps]) -> Callable[[RequiresContext[T, NewDeps]], RequiresContext[T, Deps]]:
    """Run computation with modified local context"""

Usage example:

from returns.pointfree import modify_env
from returns.context import RequiresContext

class Config:
    def __init__(self, debug: bool, db_url: str):
        self.debug = debug
        self.db_url = db_url

def enable_debug(config: Config) -> Config:
    return Config(True, config.db_url)

def get_db_connection() -> RequiresContext[str, Config]:
    return RequiresContext(lambda cfg: f"Connected to {cfg.db_url} (debug={cfg.debug})")

# Modify context before operation
debug_connection = flow(
    get_db_connection(),
    modify_env(enable_debug)
)

config = Config(False, "postgresql://localhost")
result = debug_connection(config)  # "Connected to postgresql://localhost (debug=True)"

Composition Utilities

Advanced composition functions for complex pipelines.

def unify(container: Container[T]) -> Container[T]:
    """Unify container types for composition"""

def compose_result(*functions: Callable) -> Callable:
    """Compose functions that return Results"""

def rescue(func: Callable[[E], Container[T, F]]) -> Callable[[Container[T, E]], Container[T, F]]:
    """Recover from errors with alternative computation"""

Usage examples:

from returns.pointfree import compose_result, rescue
from returns.result import Success, Failure, safe

# Compose Result-returning functions
@safe
def parse_int(s: str) -> int:
    return int(s)

@safe
def square(x: int) -> int:
    return x * x

@safe
def to_string(x: int) -> str:
    return str(x)

process_pipeline = compose_result(parse_int, square, to_string)
result = process_pipeline("5")  # Success("25")

# Error recovery
def handle_parse_error(error: Exception) -> Result[int, str]:
    return Success(0) if "invalid literal" in str(error) else Failure("Unexpected error")

safe_parse = flow(
    safe(int)("invalid"),
    rescue(handle_parse_error)
)  # Success(0)

Container-specific Operations

Specialized operations for specific container types.

# Maybe-specific
def or_else(default: Maybe[T]) -> Callable[[Maybe[T]], Maybe[T]]:
    """Provide alternative Maybe if Nothing"""

def or_else_call(func: Callable[[], Maybe[T]]) -> Callable[[Maybe[T]], Maybe[T]]:
    """Provide alternative Maybe from function if Nothing"""

# Result-specific  
def value_or(default: T) -> Callable[[Result[T, E]], T]:
    """Extract value or return default"""

def unwrap_or(default: T) -> Callable[[Container[T]], T]:
    """Unwrap container value or return default"""

# IO-specific
def unsafe_perform_io() -> Callable[[IO[T]], T]:
    """Execute IO operation (unsafe)"""

Usage examples:

from returns.pointfree import or_else, value_or, unwrap_or
from returns.maybe import Some, Nothing
from returns.result import Success, Failure

# Maybe alternatives
maybe_value = Nothing
default_some = Some("default")

result = or_else(default_some)(maybe_value)  # Some("default")

# Result value extraction
success_result = Success(42)
failure_result = Failure("error")

get_value = value_or(0)
value1 = get_value(success_result)  # 42
value2 = get_value(failure_result)   # 0

# Universal unwrap
unwrap_safe = unwrap_or("fallback")
value3 = unwrap_safe(Success("hello"))  # "hello"
value4 = unwrap_safe(Failure("error"))  # "fallback"

Point-free Composition Patterns

Pipeline Construction

from returns.pointfree import bind, map_, alt
from returns.pipeline import flow
from returns.result import Success, safe

# Build processing pipeline
@safe
def parse_float(s: str) -> float:
    return float(s)

def format_currency(amount: float) -> str:
    return f"${amount:.2f}"

def handle_error(error: Exception) -> str:
    return "Invalid amount"

process_amount = flow(
    bind(parse_float),
    map_(format_currency),
    alt(lambda _: "Invalid amount")
)

result = process_amount(Success("123.456"))  # Success("$123.46")

Function Composition

from returns.pointfree import bind, map_
from returns.curry import curry

@curry
def add(a: int, b: int) -> int:
    return a + b

@curry
def multiply(a: int, b: int) -> int:
    return a * b

# Point-free arithmetic pipeline
add_five = add(5)
double = multiply(2)

arithmetic = flow(
    map_(add_five),
    map_(double)
)

result = arithmetic(Success(10))  # Success(30)

Error Handling Pipeline

from returns.pointfree import lash, alt, rescue
from returns.result import Failure, Success

def retry_once(error: str) -> Result[int, str]:
    return Success(42) if "retry" in error else Failure("Failed permanently")

def log_error(error: str) -> str:
    print(f"Error occurred: {error}")
    return error

error_pipeline = flow(
    lash(retry_once),          # Try to recover
    alt(log_error)             # Log if still failing
)

result = error_pipeline(Failure("retry please"))  # Success(42)

Point-free operations enable clean, composable functional programming by removing the need to explicitly name intermediate values, leading to more readable and maintainable code pipelines.

Install with Tessl CLI