tessl/pypi-returns
Functional programming library providing type-safe containers for error handling, side effects, and composable operations with monadic patterns.
—
Pending
iteration-utilities.mddocs/
Iteration Utilities
Utilities for working with iterables of containers, providing declarative approaches to collection processing with type-safe error propagation and functional composition patterns.
Capabilities
Fold Operations
Declarative processing of iterables containing containers, enabling functional-style iteration without explicit loops.
class AbstractFold:
"""Abstract base for folding operations"""
class Fold:
"""Concrete implementation for declarative iterable actions"""
@staticmethod
def loop(
iterable: Iterable[Container[T]],
acc: Container[U],
func: Callable[[U, T], Container[U]]
) -> Container[U]:
"""Declarative loops for applicative types"""
@staticmethod
def collect(
iterable: Iterable[Container[T]],
acc: Container[list[T]]
) -> Container[list[T]]:
"""Transform iterable of containers to single container of list"""
@staticmethod
def collect_all(
iterable: Iterable[Container[T, E]],
acc: Container[list[T], E]
) -> Container[list[Result[T, E]], Never]:
"""Collect all values even if some operations fail"""Usage examples:
from returns.iterables import Fold
from returns.result import Success, Failure, Result
from returns.maybe import Some, Nothing, Maybe
# Collect successful results
results = [Success(1), Success(2), Success(3)]
collected = Fold.collect(results, Success([])) # Success([1, 2, 3])
# Collect with failures (fails fast)
mixed_results = [Success(1), Failure("error"), Success(3)]
collected = Fold.collect(mixed_results, Success([])) # Failure("error")
# Collect all results (including failures)
all_collected = Fold.collect_all(mixed_results, Success([]))
# Success([Success(1), Failure("error"), Success(3)])
# Loop with accumulator
def sum_values(acc: int, value: int) -> Result[int, str]:
return Success(acc + value)
numbers = [Success(1), Success(2), Success(3)]
total = Fold.loop(numbers, Success(0), sum_values) # Success(6)
# Loop with Maybe values
maybe_values = [Some(1), Some(2), Nothing, Some(4)]
def sum_maybe(acc: int, value: int) -> Maybe[int]:
return Some(acc + value)
total_maybe = Fold.loop(maybe_values, Some(0), sum_maybe) # Nothing (due to Nothing in list)Collection Processing Functions
Utility functions for processing collections of containers with various strategies.
def partition(containers: Iterable[Result[T, E]]) -> tuple[list[T], list[E]]:
"""Partition Result containers into successes and failures"""
def sequence(containers: Iterable[Container[T]]) -> Container[list[T]]:
"""Convert iterable of containers to container of list (fails fast)"""
def traverse(func: Callable[[T], Container[U]], iterable: Iterable[T]) -> Container[list[U]]:
"""Map function over iterable and sequence results"""
def filter_success(containers: Iterable[Result[T, E]]) -> Iterator[T]:
"""Extract only successful values from Result containers"""
def filter_failure(containers: Iterable[Result[T, E]]) -> Iterator[E]:
"""Extract only failure values from Result containers"""
def filter_some(containers: Iterable[Maybe[T]]) -> Iterator[T]:
"""Extract only Some values from Maybe containers"""Usage examples:
from returns.iterables import partition, sequence, traverse, filter_success
from returns.result import Success, Failure, safe
from returns.maybe import Some, Nothing
# Partition results
results = [Success(1), Failure("error1"), Success(2), Failure("error2")]
successes, failures = partition(results)
# successes: [1, 2]
# failures: ["error1", "error2"]
# Sequence containers (all must succeed)
all_success = [Success(1), Success(2), Success(3)]
sequenced = sequence(all_success) # Success([1, 2, 3])
mixed = [Success(1), Failure("error"), Success(3)]
sequenced_mixed = sequence(mixed) # Failure("error")
# Traverse with function
@safe
def double(x: int) -> int:
return x * 2
numbers = [1, 2, 3, 4]
doubled = traverse(double, numbers) # Success([2, 4, 6, 8])
# Filter successful values
mixed_results = [Success(1), Failure("error"), Success(3), Success(5)]
successful_values = list(filter_success(mixed_results)) # [1, 3, 5]
# Filter Some values
maybe_values = [Some(1), Nothing, Some(3), Nothing, Some(5)]
some_values = list(filter_some(maybe_values)) # [1, 3, 5]Async Collection Processing
Utilities for processing collections of async containers.
async def async_sequence(containers: Iterable[Future[T]]) -> Future[list[T]]:
"""Sequence Future containers asynchronously"""
async def async_traverse(
func: Callable[[T], Future[U]],
iterable: Iterable[T]
) -> Future[list[U]]:
"""Async traverse with function"""
async def async_collect(
containers: Iterable[FutureResult[T, E]]
) -> FutureResult[list[T], E]:
"""Collect FutureResult containers (fails fast)"""
async def async_collect_all(
containers: Iterable[FutureResult[T, E]]
) -> FutureResult[list[Result[T, E]], Never]:
"""Collect all FutureResult containers"""Usage examples:
import asyncio
from returns.iterables import async_sequence, async_traverse, async_collect
from returns.future import Future, FutureResult, future, future_safe
# Async sequence
@future
async def fetch_data(id: int) -> str:
await asyncio.sleep(0.1)
return f"Data {id}"
futures = [fetch_data(i) for i in range(1, 4)]
all_data = await async_sequence(futures) # ["Data 1", "Data 2", "Data 3"]
# Async traverse
@future
async def process_item(item: str) -> str:
await asyncio.sleep(0.1)
return item.upper()
items = ["hello", "world", "async"]
processed = await async_traverse(process_item, items) # ["HELLO", "WORLD", "ASYNC"]
# Async collect with error handling
@future_safe
async def risky_operation(x: int) -> int:
if x < 0:
raise ValueError(f"Negative value: {x}")
await asyncio.sleep(0.1)
return x * 2
operations = [risky_operation(i) for i in [1, 2, 3]]
results = await async_collect(operations) # Success([2, 4, 6])
# With errors
operations_with_error = [risky_operation(i) for i in [1, -2, 3]]
results_with_error = await async_collect(operations_with_error) # Failure(ValueError("Negative value: -2"))Batching and Chunking
Utilities for processing large collections in batches.
def batch_process(
items: Iterable[T],
batch_size: int,
processor: Callable[[list[T]], Container[list[U]]]
) -> Container[list[U]]:
"""Process items in batches"""
async def async_batch_process(
items: Iterable[T],
batch_size: int,
processor: Callable[[list[T]], Future[list[U]]]
) -> Future[list[U]]:
"""Async batch processing"""
def chunk(iterable: Iterable[T], size: int) -> Iterator[list[T]]:
"""Split iterable into chunks of specified size"""Usage examples:
from returns.iterables import batch_process, chunk
from returns.result import Success, safe
# Batch processing
@safe
def process_batch(items: list[int]) -> list[int]:
# Simulate batch processing (e.g., database operations)
return [x * 2 for x in items]
large_dataset = list(range(100))
processed = batch_process(large_dataset, 10, process_batch)
# Success([0, 2, 4, 6, ..., 198])
# Chunking
items = list(range(10))
chunks = list(chunk(items, 3))
# [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
# Manual batch processing with chunks
def process_in_batches(items: list[int], batch_size: int) -> Result[list[int], str]:
results = []
for batch in chunk(items, batch_size):
batch_result = process_batch(batch)
if isinstance(batch_result, Failure):
return batch_result
results.extend(batch_result.unwrap())
return Success(results)Stream Processing
Utilities for lazy stream processing with containers.
def stream_map(func: Callable[[T], Container[U]], stream: Iterator[T]) -> Iterator[Container[U]]:
"""Lazy map over stream with containers"""
def stream_filter(predicate: Callable[[T], Container[bool]], stream: Iterator[T]) -> Iterator[T]:
"""Filter stream with container-returning predicate"""
def stream_take_while(predicate: Callable[[T], Container[bool]], stream: Iterator[T]) -> Iterator[T]:
"""Take elements while predicate returns Success(True)"""
def stream_fold(
func: Callable[[U, T], Container[U]],
initial: U,
stream: Iterator[T]
) -> Container[U]:
"""Fold over stream with early termination on failure"""Usage examples:
from returns.iterables import stream_map, stream_filter, stream_fold
from returns.result import Success, Failure, safe
from returns.maybe import Some, Nothing
# Stream mapping
@safe
def parse_int(s: str) -> int:
return int(s)
number_strings = ["1", "2", "invalid", "4", "5"]
parsed_stream = stream_map(parse_int, iter(number_strings))
results = list(parsed_stream)
# [Success(1), Success(2), Failure(ValueError(...)), Success(4), Success(5)]
# Stream filtering
@safe
def is_even(x: int) -> bool:
return x % 2 == 0
numbers = [1, 2, 3, 4, 5, 6]
even_stream = stream_filter(is_even, iter(numbers))
evens = list(even_stream) # [2, 4, 6]
# Stream folding with early termination
def safe_add(acc: int, x: int) -> Result[int, str]:
if x < 0:
return Failure("Negative number encountered")
return Success(acc + x)
positive_numbers = [1, 2, 3, 4, 5]
total = stream_fold(safe_add, 0, iter(positive_numbers)) # Success(15)
mixed_numbers = [1, 2, -3, 4, 5]
total_mixed = stream_fold(safe_add, 0, iter(mixed_numbers)) # Failure("Negative number encountered")Processing Patterns
Error Accumulation
from returns.iterables import Fold, partition
from returns.result import Success, Failure
def validate_all(items: list[str]) -> Result[list[int], list[str]]:
"""Validate all items and accumulate errors"""
results = [safe(int)(item) for item in items]
# Separate successes and failures
successes, failures = partition(results)
if failures:
return Failure([str(error) for error in failures])
return Success(successes)
# Usage
valid_items = ["1", "2", "3"]
result = validate_all(valid_items) # Success([1, 2, 3])
invalid_items = ["1", "invalid", "3", "also_invalid"]
result_with_errors = validate_all(invalid_items)
# Failure(["invalid literal...", "invalid literal..."])Pipeline Processing
from returns.iterables import traverse, sequence
from returns.result import safe
from returns.pipeline import flow
# Multi-step pipeline
@safe
def validate_positive(x: int) -> int:
if x <= 0:
raise ValueError("Must be positive")
return x
@safe
def square(x: int) -> int:
return x * x
def process_numbers(numbers: list[int]) -> Result[list[int], Exception]:
return flow(
numbers,
lambda nums: traverse(validate_positive, nums),
lambda container: container.bind(lambda vals: traverse(square, vals))
)
# Usage
numbers = [1, 2, 3, 4]
result = process_numbers(numbers) # Success([1, 4, 9, 16])
numbers_with_negative = [1, -2, 3]
result_with_error = process_numbers(numbers_with_negative) # Failure(ValueError("Must be positive"))Iteration utilities provide powerful abstractions for working with collections of containers, enabling functional-style processing with proper error handling and type safety while maintaining composability and expressiveness.
Install with Tessl CLI
npx tessl i tessl/pypi-returns