Functional programming library providing type-safe containers for error handling, side effects, and composable operations with monadic patterns.
—
Core functional programming utilities for composition, transformation, and pipeline construction that enable declarative programming patterns and function composition.
Essential utility functions that form the foundation of functional programming patterns.
def identity(instance: T) -> T:
"""Returns its argument unchanged"""
def compose(first: Callable[[T], U], second: Callable[[U], V]) -> Callable[[T], V]:
"""Function composition (second after first)"""
def tap(function: Callable[[T], Any]) -> Callable[[T], T]:
"""Apply function for side effects, return original value"""
def untap(function: Callable[[T], Any]) -> Callable[[T], None]:
"""Apply function for side effects, return None"""
def raise_exception(exception: Exception) -> Callable[..., NoReturn]:
"""Helper to raise exceptions as a function"""
def not_(function: Callable[..., bool]) -> Callable[..., bool]:
"""Negate boolean function results"""Usage examples:
from returns.functions import identity, compose, tap, not_
# Identity function
value = identity(42) # 42
# Function composition
def add_one(x: int) -> int:
return x + 1
def multiply_two(x: int) -> int:
return x * 2
composed = compose(add_one, multiply_two)
result = composed(5) # (5 + 1) * 2 = 12
# Tap for side effects
def log_value(x: int) -> None:
print(f"Processing: {x}")
process_with_logging = tap(log_value)
result = process_with_logging(42) # Prints "Processing: 42", returns 42
# Negation
def is_positive(x: int) -> bool:
return x > 0
is_negative_or_zero = not_(is_positive)
result = is_negative_or_zero(-5) # TrueFunctions for creating data processing pipelines that enable point-free programming style.
def flow(instance: T, *functions: Callable) -> Any:
"""Compose value with sequence of functions"""
def pipe(*functions: Callable) -> Callable:
"""Create function pipeline"""
def managed(use: Callable[[T], Container[U]], release: Callable[[T], Any]) -> Callable[[T], Container[U]]:
"""Resource management pattern"""
def is_successful(container: Container[T, E]) -> bool:
"""Check if container represents success"""Usage examples:
from returns.pipeline import flow, pipe, is_successful
from returns.result import Success, Failure
# Flow - immediate execution
result = flow(
10,
lambda x: x * 2, # 20
lambda x: x + 5, # 25
lambda x: x / 2, # 12.5
str # "12.5"
)
# Pipe - create reusable pipeline
process_number = pipe(
lambda x: x * 2,
lambda x: x + 5,
lambda x: x / 2,
str
)
result = process_number(10) # "12.5"
# Resource management
def use_file(file_path: str) -> Success[str]:
with open(file_path, 'r') as f:
return Success(f.read())
def cleanup_file(file_path: str) -> None:
print(f"Cleaning up {file_path}")
managed_file = managed(use_file, cleanup_file)
result = managed_file("data.txt")
# Success checking
success_result = Success(42)
failure_result = Failure("error")
print(is_successful(success_result)) # True
print(is_successful(failure_result)) # FalseUtilities for currying functions and partial application with proper type safety.
def partial(func: Callable, *args, **kwargs) -> Callable:
"""Typed partial application wrapper"""
def curry(function: Callable) -> Callable:
"""Decorator for currying functions"""Usage examples:
from returns.curry import partial, curry
# Partial application
def add_three_numbers(a: int, b: int, c: int) -> int:
return a + b + c
add_10_and = partial(add_three_numbers, 10)
add_10_and_5 = partial(add_10_and, 5)
result = add_10_and_5(3) # 18
# Currying
@curry
def multiply_three(a: int, b: int, c: int) -> int:
return a * b * c
# Can be called in multiple ways
result1 = multiply_three(2)(3)(4) # 24
result2 = multiply_three(2, 3)(4) # 24
result3 = multiply_three(2)(3, 4) # 24
result4 = multiply_three(2, 3, 4) # 24
# Partial application with curried function
double = multiply_three(2)
quadruple = double(2)
result = quadruple(5) # 20Utilities for converting between different container types and flattening nested structures.
def flatten(container: Container[Container[T]]) -> Container[T]:
"""Joins two nested containers together"""
def result_to_maybe(result_container: Result[T, E]) -> Maybe[T]:
"""Converts Result to Maybe"""
def maybe_to_result(maybe_container: Maybe[T], default_error: E = None) -> Result[T, E]:
"""Converts Maybe to Result"""Usage examples:
from returns.converters import flatten, result_to_maybe, maybe_to_result
from returns.result import Success, Failure
from returns.maybe import Some, Nothing
# Flatten nested containers
nested_success = Success(Success(42))
flattened = flatten(nested_success) # Success(42)
nested_maybe = Some(Some("hello"))
flattened_maybe = flatten(nested_maybe) # Some("hello")
# Result to Maybe conversion
success_result = Success(42)
maybe_from_success = result_to_maybe(success_result) # Some(42)
failure_result = Failure("error")
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, "default error") # Success(42)
nothing_value = Nothing
result_from_nothing = maybe_to_result(nothing_value, "no value") # Failure("no value")Standalone utility methods for conditional creation and container manipulation.
def cond(condition: bool, success_value: T, failure_value: E) -> Result[T, E]:
"""Conditional container creation"""
def partition(containers: Iterable[Result[T, E]]) -> tuple[list[T], list[E]]:
"""Partition containers by success/failure"""
def unwrap_or_failure(container: Container[T, E], failure_value: E) -> T | E:
"""Unwrap container or return failure value"""Usage examples:
from returns.methods import cond, partition, unwrap_or_failure
from returns.result import Success, Failure
# Conditional creation
def validate_age(age: int) -> Result[int, str]:
return cond(age >= 18, age, "Must be 18 or older")
result1 = validate_age(25) # Success(25)
result2 = validate_age(15) # Failure("Must be 18 or older")
# Partition results
results = [
Success(1),
Failure("error1"),
Success(2),
Failure("error2"),
Success(3)
]
successes, failures = partition(results)
# successes: [1, 2, 3]
# failures: ["error1", "error2"]
# Unwrap with fallback
success_container = Success(42)
value1 = unwrap_or_failure(success_container, "fallback") # 42
failure_container = Failure("error")
value2 = unwrap_or_failure(failure_container, "fallback") # "fallback"from returns.pipeline import flow
from returns.functions import compose
# Sequential processing
data = flow(
"hello world",
str.upper, # "HELLO WORLD"
lambda s: s.split(), # ["HELLO", "WORLD"]
len, # 2
lambda n: n * 10 # 20
)
# Reusable composed function
process_text = compose(
lambda s: s.split(),
len
)from returns.curry import curry
from returns.functions import compose
@curry
def add(a: int, b: int) -> int:
return a + b
@curry
def multiply(a: int, b: int) -> int:
return a * b
# Point-free composition
add_five = add(5)
double = multiply(2)
process = compose(add_five, double)
result = process(10) # (10 + 5) * 2 = 30These functional utilities provide the building blocks for creating clean, composable, and reusable code following functional programming principles while maintaining full type safety.
Install with Tessl CLI
npx tessl i tessl/pypi-returns