tessl/pypi-toolz
List processing tools and functional utilities
functoolz.mddocs/
Function Composition
Higher-order functions for composing, currying, and transforming functions. These utilities enable elegant functional programming patterns and pipeline creation with support for memoization, partial application, and function introspection.
Capabilities
Basic Function Operations
Core utilities for working with functions and applying them in various ways.
def identity(x):
"""
Identity function - returns input unchanged.
Parameters:
- x: any value
Returns:
The input value x unchanged
"""
def apply(*func_and_args, **kwargs):
"""
Apply function to arguments and return result.
Parameters:
- func_and_args: function followed by its arguments
- **kwargs: keyword arguments for function
Returns:
Result of calling func(*args, **kwargs)
"""
def do(func, x):
"""
Run function on x for side effects, return x unchanged.
Parameters:
- func: function to call for side effects
- x: value to pass to function and return
Returns:
The input value x (func result is discarded)
"""
def flip(func, a, b):
"""
Call function with arguments flipped.
Parameters:
- func: binary function to call
- a: first argument (becomes second)
- b: second argument (becomes first)
Returns:
Result of func(b, a)
"""Function Composition
Functions for combining multiple functions into pipelines and compositions.
def compose(*funcs):
"""
Compose functions to operate in series (right to left).
Parameters:
- *funcs: functions to compose
Returns:
Function that applies all input functions in reverse order
"""
def compose_left(*funcs):
"""
Compose functions left to right.
Parameters:
- *funcs: functions to compose
Returns:
Function that applies all input functions in given order
"""
def pipe(data, *funcs):
"""
Pipe value through sequence of functions (left to right).
Parameters:
- data: initial value
- *funcs: functions to apply in sequence
Returns:
Result of applying all functions to data in sequence
"""
def thread_first(val, *forms):
"""
Thread value through sequence of functions (value as first argument).
Parameters:
- val: value to thread through functions
- *forms: functions or (function, *args) tuples
Returns:
Result of threading val through all forms as first argument
"""
def thread_last(val, *forms):
"""
Thread value through sequence of functions (value as last argument).
Parameters:
- val: value to thread through functions
- *forms: functions or (function, *args) tuples
Returns:
Result of threading val through all forms as last argument
"""Function Modification
Functions that modify or enhance the behavior of other functions.
def complement(func):
"""
Convert predicate function to its logical complement.
Parameters:
- func: predicate function returning True/False
Returns:
Function that returns opposite boolean result
"""
def juxt(*funcs):
"""
Create function that calls several functions with same arguments.
Parameters:
- *funcs: functions to call with same arguments
Returns:
Function that returns tuple of results from all input functions
"""
def memoize(func, cache=None, key=None):
"""
Cache function results for faster repeated evaluation.
Parameters:
- func: function to memoize
- cache: dictionary to use for caching (optional)
- key: function to compute cache key (optional)
Returns:
Memoized version of input function
"""
def excepts(exc, func, handler=return_none):
"""
Create function with functional try/except block.
Parameters:
- exc: exception type or tuple of types to catch
- func: function to wrap with exception handling
- handler: function to call on exception (default returns None)
Returns:
Function that catches specified exceptions and calls handler
"""
def return_none(exc):
"""
Returns None regardless of input.
Used as default exception handler for excepts function.
Parameters:
- exc: exception (ignored)
Returns:
None
"""Currying & Partial Application
Support for currying and partial application of functions.
class curry:
"""
Curry a callable for partial application.
A curried function can be called with fewer arguments than required,
returning a new function that expects the remaining arguments.
"""
def __init__(self, *args, **kwargs):
"""
Create curried function.
Parameters:
- func: function to curry
- *args: optional initial arguments
- **kwargs: optional initial keyword arguments
"""
def bind(self, *args, **kwargs):
"""
Create new curry with additional bound arguments.
Parameters:
- *args: additional positional arguments to bind
- **kwargs: additional keyword arguments to bind
Returns:
New curry instance with additional bound arguments
"""
def call(self, *args, **kwargs):
"""
Call function without currying (bypass curry behavior).
Parameters:
- *args: all positional arguments
- **kwargs: all keyword arguments
Returns:
Direct result of calling underlying function
"""Function Introspection
Utilities for inspecting function signatures and argument requirements.
def num_required_args(func, sigspec=None):
"""
Number of required positional arguments for function.
Parameters:
- func: function to inspect
- sigspec: cached signature specification (optional)
Returns:
Integer number of required positional arguments
"""
def has_varargs(func, sigspec=None):
"""
Check if function accepts variable positional arguments (*args).
Parameters:
- func: function to inspect
- sigspec: cached signature specification (optional)
Returns:
True if function accepts *args
"""
def has_keywords(func, sigspec=None):
"""
Check if function accepts keyword arguments (**kwargs).
Parameters:
- func: function to inspect
- sigspec: cached signature specification (optional)
Returns:
True if function accepts **kwargs
"""
def is_valid_args(func, args, kwargs, sigspec=None):
"""
Check if func(*args, **kwargs) is valid call.
Parameters:
- func: function to check
- args: positional arguments tuple
- kwargs: keyword arguments dict
- sigspec: cached signature specification (optional)
Returns:
True if arguments are valid for function
"""
def is_partial_args(func, args, kwargs, sigspec=None):
"""
Check if partial(func, *args, **kwargs) could be valid.
Parameters:
- func: function to check
- args: positional arguments tuple
- kwargs: keyword arguments dict
- sigspec: cached signature specification (optional)
Returns:
True if partial application could be valid
"""
def is_arity(n, func, sigspec=None):
"""
Check if function takes exactly n positional arguments.
Parameters:
- n: expected number of arguments
- func: function to check
- sigspec: cached signature specification (optional)
Returns:
True if function requires exactly n positional arguments
"""Classes
class Compose:
"""Function composition class for multiple function pipeline."""
def __init__(self, funcs):
"""
Create composition of functions.
Parameters:
- funcs: sequence of functions to compose
"""
def __call__(self, *args, **kwargs):
"""
Call composed functions on arguments.
Parameters:
- *args: positional arguments for first function
- **kwargs: keyword arguments for first function
Returns:
Result of applying all composed functions
"""
class InstanceProperty:
"""Property that returns different value when accessed on class vs instance."""
def __init__(self, fget=None, fset=None, fdel=None, doc=None, classval=None):
"""
Create instance property.
Parameters:
- fget: getter function
- fset: setter function (optional)
- fdel: deleter function (optional)
- doc: docstring (optional)
- classval: value returned when accessed on class (optional)
"""Usage Examples
Function Pipeline Creation
from toolz import pipe, compose, curry
# Data processing pipeline
def clean_text(text):
return text.strip().lower()
def remove_punctuation(text):
return ''.join(c for c in text if c.isalnum() or c.isspace())
def extract_words(text):
return text.split()
# Using pipe (left to right)
text = " Hello, World! "
words = pipe(
text,
clean_text,
remove_punctuation,
extract_words
)
# ['hello', 'world']
# Using compose (right to left)
text_processor = compose(
extract_words,
remove_punctuation,
clean_text
)
words = text_processor(" Hello, World! ")
# ['hello', 'world']Currying and Partial Application
from toolz import curry
from operator import add, mul
# Create curried functions
@curry
def multiply(x, y):
return x * y
@curry
def power(base, exponent):
return base ** exponent
# Partial application
double = multiply(2)
square = power(exponent=2)
cube = power(exponent=3)
# Use in data processing
numbers = [1, 2, 3, 4, 5]
doubled = list(map(double, numbers)) # [2, 4, 6, 8, 10]
squared = list(map(square, numbers)) # [1, 4, 9, 16, 25]
cubed = list(map(cube, numbers)) # [1, 8, 27, 64, 125]Threading Operations
from toolz import thread_first, thread_last
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Thread first (data as first argument)
result = thread_first(
data,
(filter, lambda x: x % 2 == 0), # filter(lambda x: x % 2 == 0, data)
(map, lambda x: x * 2), # map(lambda x: x * 2, filtered)
list # list(mapped)
)
# [4, 8, 12, 16, 20]
# Thread last (data as last argument)
result = thread_last(
data,
(lambda seq: filter(lambda x: x % 2 == 0, seq)),
(lambda seq: map(lambda x: x * 2, seq)),
list
)
# [4, 8, 12, 16, 20]Function Composition with Memoization
from toolz import memoize, compose
import time
@memoize
def expensive_calculation(n):
time.sleep(0.1) # Simulate expensive operation
return n * n
@memoize
def another_expensive_calc(n):
time.sleep(0.1)
return n + 10
# Compose memoized functions
pipeline = compose(another_expensive_calc, expensive_calculation)
# First call is slow
start = time.time()
result1 = pipeline(5) # expensive_calculation(5) -> another_expensive_calc(25)
time1 = time.time() - start
# Second call is fast (memoized)
start = time.time()
result2 = pipeline(5) # Both results cached
time2 = time.time() - start
print(f"First call: {time1:.2f}s, Second call: {time2:.4f}s")
# First call: 0.20s, Second call: 0.0001sFunction Utilities
from toolz import juxt, complement, do
# juxt - call multiple functions with same arguments
def add_one(x): return x + 1
def multiply_two(x): return x * 2
def square(x): return x * x
multi_transform = juxt(add_one, multiply_two, square)
result = multi_transform(5) # (6, 10, 25)
# complement - logical opposite
def is_even(x): return x % 2 == 0
is_odd = complement(is_even)
evens = list(filter(is_even, [1, 2, 3, 4, 5])) # [2, 4]
odds = list(filter(is_odd, [1, 2, 3, 4, 5])) # [1, 3, 5]
# do - side effects while passing through
def log_value(x):
print(f"Processing: {x}")
process_with_logging = lambda x: do(log_value, x * 2)
result = process_with_logging(5) # Prints "Processing: 10", returns 10Install with Tessl CLI
npx tessl i tessl/pypi-toolz