When they're not builtins, they're boltons.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive tools for working with iterables including chunking, windowing, flattening, uniqueness operations, and advanced data structure manipulation. Provides both list and iterator versions of operations for memory efficiency and performance optimization.
Break iterables into chunks of specified sizes for batch processing.
def chunked(src, size, count=None, **kw):
"""
Break iterable into chunks of specified size.
Parameters:
- src (iterable): Source iterable to chunk
- size (int): Size of each chunk
- count (int, optional): Maximum number of chunks to return
Returns:
list: List of chunks (each chunk is a list)
"""
def chunked_iter(src, size, **kw):
"""
Iterator version of chunked.
Parameters:
- src (iterable): Source iterable to chunk
- size (int): Size of each chunk
Yields:
list: Each chunk as a list
"""
def chunk_ranges(input_size, chunk_size, **kwargs):
"""
Generate ranges for chunking operations.
Parameters:
- input_size (int): Total size of input data
- chunk_size (int): Desired chunk size
Returns:
list: List of (start, end) tuples representing chunk ranges
"""Create sliding windows over iterables for sequential data analysis.
def windowed(src, size, fill=_UNSET):
"""
Create sliding window over iterable.
Parameters:
- src (iterable): Source iterable
- size (int): Window size
- fill: Fill value for incomplete windows (default: no fill)
Returns:
list: List of windows (each window is a tuple)
"""
def windowed_iter(src, size, fill=_UNSET):
"""
Iterator version of windowed.
Parameters:
- src (iterable): Source iterable
- size (int): Window size
- fill: Fill value for incomplete windows
Yields:
tuple: Each window as a tuple
"""
def pairwise(src, end=_UNSET):
"""
Iterate over successive pairs from iterable.
Parameters:
- src (iterable): Source iterable
- end: Value to pair with last element (default: no end pairing)
Returns:
list: List of (current, next) tuples
"""
def pairwise_iter(src, end=_UNSET):
"""
Iterator version of pairwise.
Parameters:
- src (iterable): Source iterable
- end: Value to pair with last element
Yields:
tuple: Each (current, next) pair
"""Split, strip, and manipulate sequences with flexible separators.
def split(src, sep=None, maxsplit=None):
"""
Split iterable by separator.
Parameters:
- src (iterable): Source iterable to split
- sep: Separator value (None means any falsy value)
- maxsplit (int, optional): Maximum number of splits
Returns:
list: List of split segments
"""
def split_iter(src, sep=None, maxsplit=None):
"""Iterator version of split."""
def lstrip(iterable, strip_value=None):
"""
Strip values from left side of iterable.
Parameters:
- iterable: Source iterable
- strip_value: Value to strip (None means falsy values)
Returns:
list: Iterable with left values stripped
"""
def rstrip(iterable, strip_value=None):
"""Strip values from right side of iterable."""
def strip(iterable, strip_value=None):
"""Strip values from both sides of iterable."""
def strip_iter(iterable, strip_value=None):
"""Iterator version of strip."""Remove duplicates, find redundant elements, and filter iterables.
def unique(src, key=None):
"""
Remove duplicates while preserving order.
Parameters:
- src (iterable): Source iterable
- key (callable, optional): Function to compute unique key
Returns:
list: List with duplicates removed
"""
def unique_iter(src, key=None):
"""Iterator version of unique."""
def redundant(src, key=None, **kwargs):
"""
Find duplicate elements.
Parameters:
- src (iterable): Source iterable
- key (callable, optional): Function to compute comparison key
Returns:
list: List of duplicate elements
"""
def partition(src, key=None, **kwargs):
"""
Partition iterable into two lists by predicate.
Parameters:
- src (iterable): Source iterable
- key (callable): Predicate function
Returns:
tuple: (truthy_items, falsy_items)
"""Recursively transform and query nested data structures.
def remap(root_obj, **kwargs):
"""
Recursively transform nested data structures.
Parameters:
- root_obj: Root object to transform
- visit (callable, optional): Function to visit each value
- enter (callable, optional): Function called when entering containers
- exit (callable, optional): Function called when exiting containers
Returns:
Transformed data structure
"""
def get_path(root_obj, path, default=_UNSET):
"""
Get value from nested structure by path.
Parameters:
- root_obj: Root object to query
- path (iterable): Path to the desired value
- default: Default value if path not found
Returns:
Value at the specified path
"""
def research(root_obj, **kwargs):
"""
Search nested structures with query function.
Parameters:
- root_obj: Root object to search
- query (callable): Function to test values
Returns:
list: List of matching values
"""Recursively flatten nested iterables and data structures.
def flatten(iterable):
"""
Flatten nested iterables to list.
Parameters:
- iterable: Nested iterable to flatten
Returns:
list: Flattened list
"""
def flatten_iter(iterable):
"""
Recursively flatten nested iterables.
Parameters:
- iterable: Nested iterable to flatten
Yields:
Individual items from nested structure
"""Miscellaneous iteration utilities for common patterns.
def bucketize(src, key=None, **kwargs):
"""
Group items into buckets by key function.
Parameters:
- src (iterable): Source iterable
- key (callable): Function to compute bucket key
Returns:
dict: Dictionary mapping keys to lists of items
"""
def one(src, default=_UNSET, **kwargs):
"""
Ensure iterable contains exactly one element.
Parameters:
- src (iterable): Source iterable
- default: Default value if iterable is empty
Returns:
The single element
Raises:
ValueError: If iterable has more than one element
"""
def first(iterable, default=_UNSET, key=None):
"""
Get first element matching condition.
Parameters:
- iterable: Source iterable
- default: Default if no match found
- key (callable, optional): Condition function
Returns:
First matching element or default
"""
def same(iterable, ref=_UNSET):
"""
Check if all elements are the same.
Parameters:
- iterable: Source iterable
- ref: Reference value to compare against
Returns:
bool: True if all elements are the same
"""Extended range functions with float support and exponential backoff.
def xfrange(stop, start=None, step=1.0):
"""
Extended float range generator.
Parameters:
- stop (float): End value (exclusive)
- start (float, optional): Start value (default: 0)
- step (float): Step size (default: 1.0)
Yields:
float: Each value in the range
"""
def frange(stop, start=None, step=1.0):
"""
Float range as list.
Parameters:
- stop (float): End value (exclusive)
- start (float, optional): Start value
- step (float): Step size
Returns:
list: List of float values
"""
def backoff(start=1, **kwargs):
"""
Generate exponential backoff delays.
Parameters:
- start (float): Initial delay value
- factor (float): Multiplier for each step (default: 2)
- max (float): Maximum delay value
- jitter (bool): Add random jitter
Returns:
list: List of delay values
"""
def backoff_iter(start=1, **kwargs):
"""Iterator version of backoff."""Sort operations with error handling for mixed types.
def soft_sorted(iterable, **kwargs):
"""
Sort with soft error handling for incomparable types.
Parameters:
- iterable: Source iterable to sort
- key (callable, optional): Function to compute sort key
- reverse (bool): Sort in reverse order
Returns:
list: Sorted list with error handling
"""
def untyped_sorted(iterable, **kwargs):
"""
Sort mixed types by converting to strings.
Parameters:
- iterable: Source iterable with mixed types
- key (callable, optional): Function to compute sort key
- reverse (bool): Sort in reverse order
Returns:
list: Sorted list with all types converted to strings
"""Utilities for checking iterable properties.
def is_iterable(obj):
"""
Check if object is iterable.
Parameters:
- obj: Object to check
Returns:
bool: True if object is iterable
"""
def is_scalar(obj):
"""
Check if object is scalar (non-iterable).
Parameters:
- obj: Object to check
Returns:
bool: True if object is scalar
"""
def is_collection(obj):
"""
Check if object is collection (iterable but not string/bytes).
Parameters:
- obj: Object to check
Returns:
bool: True if object is a collection
"""from boltons.iterutils import (
chunked, windowed, flatten, unique, remap,
get_path, backoff, soft_sorted
)
# Process data in chunks
data = list(range(100))
for chunk in chunked(data, 10):
print(f"Processing {len(chunk)} items")
# Create sliding windows
sequence = [1, 2, 3, 4, 5]
windows = windowed(sequence, 3)
print(windows) # [(1, 2, 3), (2, 3, 4), (3, 4, 5)]
# Flatten nested structures
nested = [[1, 2], [3, [4, 5]], 6]
flat = flatten(nested)
print(flat) # [1, 2, 3, 4, 5, 6]
# Remove duplicates preserving order
items = [1, 2, 2, 3, 1, 4]
unique_items = unique(items)
print(unique_items) # [1, 2, 3, 4]
# Transform nested data structures
data = {'a': [1, 2], 'b': {'c': 3}}
doubled = remap(data, visit=lambda path, key, value:
value * 2 if isinstance(value, int) else value)
print(doubled) # {'a': [2, 4], 'b': {'c': 6}}
# Extract values by path
nested_dict = {'user': {'profile': {'name': 'Alice'}}}
name = get_path(nested_dict, ['user', 'profile', 'name'])
print(name) # 'Alice'
# Generate exponential backoff delays
delays = backoff(start=0.1, factor=2, max=10)
print(delays) # [0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 10, 10, ...]class PathAccessError(Exception):
"""Exception for invalid path access in nested structures."""
pass
class GUIDerator:
"""Base iterator with guidance functionality."""
def __init__(self, iterable): ...
def __iter__(self): ...
def __next__(self): ...
class SequentialGUIDerator(GUIDerator):
"""Sequential iterator with reseed capability."""
def reseed(self, new_iterable): ...
# Sentinel values
_UNSET = object() # Sentinel for unset valuesInstall with Tessl CLI
npx tessl i tessl/pypi-boltons