Collections Module¶

The rite.collections module provides advanced data structures including caches, buffers, trees, and collection utilities.

Overview¶

collections ¶

Collections Module¶

Comprehensive data structures and utilities for common programming patterns, organized into semantic submodules.

Submodules¶

list: List manipulation (unique, flatten, chunk, group, partition)
dict: Dictionary operations (merge, filter, invert, deep access)
set: Set operations (union, intersection, difference)
buffer: Ring buffers, circular buffers, and sliding windows
cache: LRU, LFU, and TTL caches
queue: Priority queues, circular queues, and deque wrappers
tree: Tree nodes, binary trees, tries, and nested sets
pattern: Design patterns (singleton, observer, object pool)

Examples¶

from rite.collections import list_unique, dict_merge list_unique([1, 2, 2, 3]) [1, 2, 3] dict_merge({"a": 1}, {"b": 2})

from rite.collections import CircularBuffer, LRUCache buf = CircularBuffer(5) cache = LRUCache(100)

Classes¶

BinaryTreeNode ¶

BinaryTreeNode(value: Any, left: BinaryTreeNode | None = None, right: BinaryTreeNode | None = None)

BinaryTreeNode Class¶

A binary tree node with left and right children.

Parameters¶

value : Any The value stored in this node. left : BinaryTreeNode | None Left child node. right : BinaryTreeNode | None Right child node.

Initialize a binary tree node.

value: The value to store.
left: Left child node.
right: Right child node.

Functions¶

get_height ¶

get_height() -> int

Get height of subtree rooted at this node.

int: Height of subtree.

has_left_child ¶

has_left_child() -> bool

Check if node has left child.

bool: True if left child exists.

has_right_child ¶

has_right_child() -> bool

Check if node has right child.

bool: True if right child exists.

inorder_traversal ¶

inorder_traversal() -> list[BinaryTreeNode]

Perform in-order traversal (left, root, right).

list[Any]: Values in in-order.

is_leaf ¶

is_leaf() -> bool

Check if node is a leaf.

bool: True if node has no children.

postorder_traversal ¶

postorder_traversal() -> list[BinaryTreeNode]

Perform post-order traversal (left, right, root).

list[Any]: Values in post-order.

preorder_traversal ¶

preorder_traversal() -> list[BinaryTreeNode]

Perform pre-order traversal (root, left, right).

list[Any]: Values in pre-order.

traverse_inorder ¶

traverse_inorder() -> list[BinaryTreeNode]

Alias for inorder_traversal for compatibility.

traverse_postorder ¶

traverse_postorder() -> list[BinaryTreeNode]

Alias for postorder_traversal for compatibility.

traverse_preorder ¶

traverse_preorder() -> list[BinaryTreeNode]

Alias for preorder_traversal for compatibility.

BoundedBuffer ¶

BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')

BoundedBuffer Class¶

A size-limited buffer that prevents overflow with configurable behavior.

Parameters¶

maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'

Initialize the bounded buffer.

maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.

Attributes¶

capacity `property` ¶

capacity: int

Return maximum number of elements the buffer can hold.

Functions¶

append ¶

append(item: Any) -> bool

Add an item to the buffer.

item: The item to add.

bool: True if item was added, False if dropped.

BufferError: If overflow_strategy is 'raise' and buffer is full.

clear ¶

clear() -> None

Clear all items from buffer.

extend ¶

extend(items: list[Any]) -> None

Extend buffer with multiple items.

items: List of items to add.

get ¶

get(index: int) -> Any | None

Return the item at a given index or None if out of bounds.

get_all ¶

get_all() -> list[Any]

Get all items in buffer.

list[Any]: All items in insertion order.

is_empty ¶

is_empty() -> bool

Check if buffer is empty.

bool: True if buffer contains no items.

is_full ¶

is_full() -> bool

Check if buffer is full.

bool: True if buffer has reached maxsize.

peek ¶

peek() -> Any | None

Return the oldest item without removing it.

CircularBuffer ¶

CircularBuffer(size: int)

CircularBuffer Class¶

A fixed-size buffer that overwrites the oldest data when full.

Attributes¶

size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.

Methods¶

append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.

Initializes a CircularBuffer instance.

Parameters:¶

size : int The maximum number of elements the buffer can hold.

Functions¶

append ¶

append(value: Any) -> None

Adds a value to the buffer. Overwrites the oldest element if the buffer is full.

Parameters:¶

value : Any The value to add to the buffer.

clear ¶

clear() -> None

Clear the buffer.

get ¶

get(index: int) -> Any | None

Return item at index or None if out of bounds.

get_all ¶

get_all() -> list[Any | None]

Retrieves all elements in the buffer in the correct order.

Returns¶

list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.

is_empty ¶

is_empty() -> bool

Checks if the buffer is empty.

Returns¶

bool: True if the buffer is empty, False otherwise.

is_full ¶

is_full() -> bool

Checks if the buffer is full.

Returns¶

bool: True if the buffer is full, False otherwise.

CircularQueue ¶

CircularQueue(capacity: int)

CircularQueue Class¶

A fixed-size circular queue (FIFO).

Initialize a circular queue.

capacity: Maximum capacity of the queue.

Functions¶

clear ¶

clear() -> None

Clear all items from queue.

dequeue ¶

dequeue() -> Any

Remove and return item from front of queue.

Any: The front item.

enqueue ¶

enqueue(item: Any) -> bool

Add an item to the rear of the queue.

item: Item to add.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

is_full ¶

is_full() -> bool

Check if queue is full.

peek ¶

peek() -> Any

Return front item without removing it.

Any: The front item.

DequeWrapper ¶

DequeWrapper(max_size: int | None = None)

DequeWrapper Class¶

A wrapper around collections.deque with additional utilities.

Initialize a deque wrapper.

max_size: Maximum length of the deque.

Functions¶

clear ¶

clear() -> None

Remove all items.

is_empty ¶

is_empty() -> bool

Check if deque is empty.

peek_left ¶

peek_left() -> Any

Return item at left end without removing.

peek_right ¶

peek_right() -> Any

Return item at right end without removing.

pop_left ¶

pop_left() -> Any

Remove and return item from left end.

pop_right ¶

pop_right() -> Any

Remove and return item from right end.

push_left ¶

push_left(item: Any) -> None

Add item to the left end.

push_right ¶

push_right(item: Any) -> None

Add item to the right end.

rotate ¶

rotate(n: int = 1) -> None

Rotate deque n steps to the right.

n: Number of steps to rotate (negative for left rotation).

to_list ¶

to_list() -> list[Any]

Convert to list.

LFUCache ¶

LFUCache(capacity: int)

LFUCache Class¶

A Least Frequently Used (LFU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LFU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

LRUCache ¶

LRUCache(capacity: int)

LRUCache Class¶

A Least Recently Used (LRU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LRU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

items ¶

items() -> list[tuple[Any, Any]]

Get all key-value pairs.

list[tuple[Any, Any]]: List of (key, value) tuples.

keys ¶

keys() -> list[Any]

Get all keys in cache.

list[Any]: List of keys in order of use.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

values ¶

values() -> list[Any]

Get all values in cache.

list[Any]: List of values.

NestedSetStructure ¶

NestedSetStructure()

Nested Set Structure Class¶

A structure to track items and their hierarchical relationships. Supports multi-level trees with parent/child tracking. Useful for scenarios like recursive publishing, dependency resolution, or graph-like traversal.

Initialize the nested set structure.

Functions¶

add ¶

add(item: Any, parent: Any | None = None) -> None

Add an item to the structure, optionally specifying a parent.

children ¶

children(item: Any) -> list[Any]

Return the children of the given item. If the item has no children, returns an empty list.

nested_items ¶

nested_items() -> list[Any]

Return a nested, flattened list of items (including children) in insertion order. Maintains hierarchy order.

original ¶

original(item: Any) -> Any

Return the original object stored in the set that equals item, or item itself if not found. This is useful for matching back to canonical instances.

parent ¶

parent(item: Any) -> Any | None

Return the parent of the given item, or None if it has no parent.

ObjectPool ¶

ObjectPool(factory: Callable[[], Any], max_size: int = 10, reset: Callable[[Any], None] | None = None)

ObjectPool Class¶

A generic object pool for managing reusable objects.

Parameters¶

factory : Callable[[], Any] Factory function to create new objects. max_size : int Maximum number of objects in the pool. reset_func : Callable[[Any], None] | None Optional function to reset objects before reuse.

Initialize an object pool.

factory: Function to create new objects.
max_size: Maximum pool size.
reset_func: Optional function to reset objects.

Functions¶

acquire ¶

acquire() -> Any

Acquire an object from the pool.

Any: An object from the pool or newly created.

available_count ¶

available_count() -> int

Get number of available objects.

int: Number of available objects.

clear ¶

clear() -> None

Clear all available objects from the pool.

in_use_count ¶

in_use_count() -> int

Get number of objects currently in use.

int: Number of in-use objects.

release ¶

release(obj: Any) -> None

Release an object back to the pool.

obj: The object to release.

ValueError: If object was not acquired from this pool.

size ¶

size() -> int

Get total number of objects managed by pool.

int: Total objects (available + in use).

Observable ¶

Observable()

Observable Class¶

Base class for observable objects in the Observer pattern.

Initialize an observable object.

Attributes¶

observers `property` ¶

observers: tuple[Observer, ...]

Return observers as an immutable tuple.

Functions¶

attach ¶

attach(observer: Observer) -> None

Attach an observer.

observer: The observer to attach.

clear_observers ¶

clear_observers() -> None

Remove all observers.

detach ¶

detach(observer: Observer) -> None

Detach an observer.

observer: The observer to detach.

get_observer_count ¶

get_observer_count() -> int

Get the number of attached observers.

int: Number of observers.

notify ¶

notify(*args: Any, **kwargs: Any) -> None

Notify all observers of a change.

*args: Positional arguments to pass to observers.
**kwargs: Keyword arguments to pass to observers.

Observer ¶

Observer Abstract Base Class¶

Base class for observers in the Observer pattern.

Functions¶

update ¶

update(observable: Observable | None, *args: Any, **kwargs: Any) -> None

Called when the observed object changes.

PriorityQueue ¶

PriorityQueue()

PriorityQueue Class¶

A priority queue where items with lower priority values are dequeued first.

Initialize an empty priority queue.

Functions¶

clear ¶

clear() -> None

Remove all items from the queue.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

bool: True if queue is empty.

peek ¶

peek() -> Any

Return the highest priority item without removing it.

Any: The highest priority item.

pop ¶

pop() -> Any

Remove and return the highest priority item.

Any: The highest priority item.

push ¶

push(item: Any, priority: float = 0.0) -> None

Add an item with a given priority.

item: The item to add.
priority: Priority value (lower = higher priority).

SingletonMeta ¶

Bases: type

SingletonMeta Class¶

A metaclass for implementing the Singleton pattern. This ensures that only one instance of a class exists throughout the application lifecycle.

Attributes¶

_instances : dict[type, Any] A dictionary that maps classes to their single instances.

Methods¶

call(*args, **kwargs): Overrides the __call__ method to control instance creation.

Example¶

# Define a Singleton class
class Configuration(metaclass=SingletonMeta):
    def __init__(self, value):
        self.value = value

Functions¶

reset_instance `classmethod` ¶

reset_instance(target_cls: type) -> None

Remove a cached instance for the given class if present.

SlidingWindow ¶

SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)

SlidingWindow Class¶

A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.

Parameters¶

size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.

Initialize the sliding window.

size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.

Functions¶

add ¶

add(value: Any) -> Any | None

Add a value to the window and return aggregated result.

value: The value to add to the window.

Any | None: Aggregated result if aggregation_func is set.

clear ¶

clear() -> None

Clear all elements from window.

get_aggregate ¶

get_aggregate() -> Any | None

Get aggregated value of current window.

Any | None: Aggregated result or None if no aggregation function.

get_window ¶

get_window() -> list[Any]

Get current window contents.

list[Any]: Current window values in order.

is_empty ¶

is_empty() -> bool

Check if window is empty.

bool: True if window contains no elements.

is_full ¶

is_full() -> bool

Check if window is full.

bool: True if window has reached its size.

moving_average ¶

moving_average() -> float | None

Calculate moving average of numeric window.

float | None: Average of window values or None if empty.

moving_max ¶

moving_max() -> Any | None

Get maximum value in window.

Any | None: Maximum value or None if empty.

moving_min ¶

moving_min() -> Any | None

Get minimum value in window.

Any | None: Minimum value or None if empty.

moving_sum ¶

moving_sum() -> float

Calculate moving sum of numeric window.

float | None: Sum of window values or None if empty.

TTLCache ¶

TTLCache(default_ttl: float = 300.0, max_size: int | None = None)

TTLCache Class¶

A cache with Time-To-Live (TTL) expiration for entries.

Parameters¶

default_ttl : float Default time-to-live in seconds. max_size : int | None Optional maximum cache size.

Initialize a TTL cache.

default_ttl: Default expiration time in seconds.
max_size: Optional maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache if not expired.

key: The key to look up.

Any | None: The value or None if not found/expired.

get_ttl ¶

get_ttl(key: Any) -> float | None

Get remaining TTL for a key.

key: The key to check.

float | None: Remaining seconds or None if not found/expired.

put ¶

put(key: Any, value: Any, ttl: float | None = None) -> None

Add or update a key-value pair with TTL.

key: The key to store.
value: The value to store.
ttl: Optional TTL in seconds (uses default if not provided).

TreeNode ¶

TreeNode(value: Any, children: list[Self] | None = None)

TreeNode Class¶

A generic tree node that can have multiple children.

Parameters¶

value : Any The value stored in this node. children : list[TreeNode] | None Optional list of child nodes.

Initialize a tree node.

value: The value to store in this node.
children: Optional list of child nodes.

Functions¶

add_child ¶

add_child(child: Self) -> None

Add a child node.

child: The child node to add.

get_ancestors ¶

get_ancestors() -> list[Self]

Get all ancestor nodes from parent to root.

list[TreeNode]: List of ancestors in order (parent to root).

get_depth ¶

get_depth() -> int

Get the depth of this node (distance from root).

int: Depth of node (root is 0).

get_height ¶

get_height() -> int

Get the height of the subtree rooted at this node.

int: Height of subtree (leaf is 0).

get_siblings ¶

get_siblings() -> list[Self]

Get all sibling nodes (nodes with same parent).

list[TreeNode]: List of sibling nodes.

is_leaf ¶

is_leaf() -> bool

Check if this node is a leaf (has no children).

bool: True if node has no children.

is_root ¶

is_root() -> bool

Check if this node is a root (has no parent).

bool: True if node has no parent.

remove_child ¶

remove_child(child: Self) -> bool

Remove a child node.

child: The child node to remove.

bool: True if child was removed, False if not found.

traverse_levelorder ¶

traverse_levelorder() -> list[Self]

Traverse tree in level-order (breadth-first).

list[TreeNode]: Nodes in level-order.

traverse_postorder ¶

traverse_postorder() -> list[Self]

Traverse tree in post-order (children, then root).

list[TreeNode]: Nodes in post-order.

traverse_preorder ¶

traverse_preorder() -> list[Self]

Traverse tree in pre-order (root, then children).

list[TreeNode]: Nodes in pre-order.

Trie ¶

Trie()

Trie Class¶

A prefix tree (trie) for efficient string operations like autocomplete, spell checking, and prefix matching.

Initialize an empty trie.

Functions¶

autocomplete ¶

autocomplete(prefix: str) -> list[str]

Return all completions for the given prefix.

delete ¶

delete(word: str) -> bool

Delete a word from the trie.

word: The word to delete.

bool: True if word was deleted, False if not found.

get ¶

get(word: str) -> Any | None

Get the value associated with a word.

word: The word to look up.

Any | None: Associated value or None if not found.

get_all_words ¶

get_all_words() -> list[str]

Get all words stored in the trie.

list[str]: All words in the trie.

get_words_with_prefix ¶

get_words_with_prefix(prefix: str) -> list[str]

Get all words that start with given prefix.

prefix: The prefix to match.

list[str]: List of words with the prefix.

insert ¶

insert(word: str, value: Any | None = None) -> None

Insert a word into the trie.

word: The word to insert.
value: Optional value to associate with the word.

search ¶

search(word: str) -> bool

Search for an exact word in the trie.

word: The word to search for.

bool: True if word exists in trie.

starts_with ¶

starts_with(prefix: str) -> bool

Check if any word in trie starts with given prefix.

prefix: The prefix to check.

bool: True if prefix exists in trie.

Functions¶

set_difference ¶

set_difference(first: set[T], *others: set[T]) -> set[T]

Return difference of first set minus all others.

Parameters:

Name	Type	Description	Default
`first`	`set[T]`	Base set to subtract from.	required
`*others`	`set[T]`	Sets to subtract.	`()`

Returns:

Type	Description
`set[T]`	New set with elements in first but not in others.

Examples:

>>> set_difference({1, 2, 3}, {2}, {3})
{1}
>>> set_difference({1, 2, 3, 4}, {2, 3})
{1, 4}

set_intersection ¶

set_intersection(*sets: set[T]) -> set[T]

Return intersection of multiple sets.

Parameters:

Name	Type	Description	Default
`*sets`	`set[T]`	Variable number of sets to intersect.	`()`

Returns:

Type	Description
`set[T]`	New set containing only common elements.

Examples:

>>> set_intersection({1, 2, 3}, {2, 3, 4}, {2, 3, 5})
{2, 3}
>>> set_intersection({1, 2}, {3, 4})
set()

set_symmetric_difference ¶

set_symmetric_difference(first: set[T], second: set[T]) -> set[T]

Return symmetric difference (elements in either but not both).

Parameters:

Name	Type	Description	Default
`first`	`set[T]`	First set.	required
`second`	`set[T]`	Second set.	required

Returns:

Type	Description
`set[T]`	New set with elements in first or second but not both.

Examples:

>>> set_symmetric_difference({1, 2, 3}, {2, 3, 4})
{1, 4}
>>> set_symmetric_difference({1, 2}, {3, 4})
{1, 2, 3, 4}

set_union ¶

set_union(*sets: set[T]) -> set[T]

Return union of multiple sets.

Parameters:

Name	Type	Description	Default
`*sets`	`set[T]`	Variable number of sets to unite.	`()`

Returns:

Type	Description
`set[T]`	New set containing all unique elements.

Examples:

>>> set_union({1, 2}, {2, 3}, {3, 4})
{1, 2, 3, 4}
>>> set_union({1}, {2}, {3})
{1, 2, 3}

Modules¶

buffer ¶

Buffer Collections Module¶

Provides various buffer data structures for specialized data storage patterns.

Classes:¶

CircularBuffer: Ring buffer with automatic overwriting
RingBuffer: Alias for CircularBuffer
BoundedBuffer: Size-limited buffer with overflow protection
SlidingWindow: Moving window over a data stream

Classes¶

BoundedBuffer ¶

BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')

BoundedBuffer Class¶

A size-limited buffer that prevents overflow with configurable behavior.

Parameters¶

maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'

Initialize the bounded buffer.

maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.

Attributes¶

capacity property ¶

capacity: int

Return maximum number of elements the buffer can hold.

Functions¶

append ¶

append(item: Any) -> bool

Add an item to the buffer.

item: The item to add.

bool: True if item was added, False if dropped.

BufferError: If overflow_strategy is 'raise' and buffer is full.

clear ¶

clear() -> None

Clear all items from buffer.

extend ¶

extend(items: list[Any]) -> None

Extend buffer with multiple items.

items: List of items to add.

get ¶

get(index: int) -> Any | None

Return the item at a given index or None if out of bounds.

get_all ¶

get_all() -> list[Any]

Get all items in buffer.

list[Any]: All items in insertion order.

is_empty ¶

is_empty() -> bool

Check if buffer is empty.

bool: True if buffer contains no items.

is_full ¶

is_full() -> bool

Check if buffer is full.

bool: True if buffer has reached maxsize.

peek ¶

peek() -> Any | None

Return the oldest item without removing it.

CircularBuffer ¶

CircularBuffer(size: int)

CircularBuffer Class¶

A fixed-size buffer that overwrites the oldest data when full.

Attributes¶

size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.

Methods¶

append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.

Initializes a CircularBuffer instance.

Parameters:¶

size : int The maximum number of elements the buffer can hold.

Functions¶

append ¶

append(value: Any) -> None

Adds a value to the buffer. Overwrites the oldest element if the buffer is full.

Parameters:¶

value : Any The value to add to the buffer.

clear ¶

clear() -> None

Clear the buffer.

get ¶

get(index: int) -> Any | None

Return item at index or None if out of bounds.

get_all ¶

get_all() -> list[Any | None]

Retrieves all elements in the buffer in the correct order.

Returns¶

list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.

is_empty ¶

is_empty() -> bool

Checks if the buffer is empty.

Returns¶

bool: True if the buffer is empty, False otherwise.

is_full ¶

is_full() -> bool

Checks if the buffer is full.

Returns¶

bool: True if the buffer is full, False otherwise.

SlidingWindow ¶

SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)

SlidingWindow Class¶

A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.

Parameters¶

size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.

Initialize the sliding window.

size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.

Functions¶

add ¶

add(value: Any) -> Any | None

Add a value to the window and return aggregated result.

value: The value to add to the window.

Any | None: Aggregated result if aggregation_func is set.

clear ¶

clear() -> None

Clear all elements from window.

get_aggregate ¶

get_aggregate() -> Any | None

Get aggregated value of current window.

Any | None: Aggregated result or None if no aggregation function.

get_window ¶

get_window() -> list[Any]

Get current window contents.

list[Any]: Current window values in order.

is_empty ¶

is_empty() -> bool

Check if window is empty.

bool: True if window contains no elements.

is_full ¶

is_full() -> bool

Check if window is full.

bool: True if window has reached its size.

moving_average ¶

moving_average() -> float | None

Calculate moving average of numeric window.

float | None: Average of window values or None if empty.

moving_max ¶

moving_max() -> Any | None

Get maximum value in window.

Any | None: Maximum value or None if empty.

moving_min ¶

moving_min() -> Any | None

Get minimum value in window.

Any | None: Minimum value or None if empty.

moving_sum ¶

moving_sum() -> float

Calculate moving sum of numeric window.

float | None: Sum of window values or None if empty.

Modules¶

bounded_buffer ¶

Bounded Buffer¶

A thread-safe bounded buffer that blocks or raises errors when capacity limits are reached.

Classes¶

BoundedBuffer ¶

BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')

BoundedBuffer Class¶

A size-limited buffer that prevents overflow with configurable behavior.

Parameters¶

maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'

Initialize the bounded buffer.

maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.

Attributes¶

capacity property ¶

capacity: int

Return maximum number of elements the buffer can hold.

Functions¶

append ¶

append(item: Any) -> bool

Add an item to the buffer.

item: The item to add.

bool: True if item was added, False if dropped.

BufferError: If overflow_strategy is 'raise' and buffer is full.

clear ¶

clear() -> None

Clear all items from buffer.

extend ¶

extend(items: list[Any]) -> None

Extend buffer with multiple items.

items: List of items to add.

get ¶

get(index: int) -> Any | None

Return the item at a given index or None if out of bounds.

get_all ¶

get_all() -> list[Any]

Get all items in buffer.

list[Any]: All items in insertion order.

is_empty ¶

is_empty() -> bool

Check if buffer is empty.

bool: True if buffer contains no items.

is_full ¶

is_full() -> bool

Check if buffer is full.

bool: True if buffer has reached maxsize.

peek ¶

peek() -> Any | None

Return the oldest item without removing it.

circular_buffer ¶

Circular Buffer¶

A circular buffer (ring buffer) implementation that overwrites the oldest entries when the buffer reaches its maximum capacity.

Classes:¶

CircularBuffer: A fixed-size buffer with circular indexing.

Features:¶

Append new elements, overwriting the oldest if the buffer is full.
Retrieve all elements in the correct order.
Check if the buffer is empty or full.

Usage:¶

buffer = CircularBuffer(size=5)
buffer.append(1)
buffer.append(2)
print(buffer.get_all())  # Outputs: [1, 2]
buffer.append(3)
buffer.append(4)
buffer.append(5)
print(buffer.get_all())  # Outputs: [1, 2, 3, 4, 5]
buffer.append(6)
print(buffer.get_all())  # Outputs: [6, 2, 3, 4, 5]

Classes¶

CircularBuffer ¶

CircularBuffer(size: int)

CircularBuffer Class¶

A fixed-size buffer that overwrites the oldest data when full.

Attributes¶

size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.

Methods¶

append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.

Initializes a CircularBuffer instance.

Parameters:¶

size : int The maximum number of elements the buffer can hold.

Functions¶

append ¶

append(value: Any) -> None

Adds a value to the buffer. Overwrites the oldest element if the buffer is full.

Parameters:¶

value : Any The value to add to the buffer.

clear ¶

clear() -> None

Clear the buffer.

get ¶

get(index: int) -> Any | None

Return item at index or None if out of bounds.

get_all ¶

get_all() -> list[Any | None]

Retrieves all elements in the buffer in the correct order.

Returns¶

list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.

is_empty ¶

is_empty() -> bool

Checks if the buffer is empty.

Returns¶

bool: True if the buffer is empty, False otherwise.

is_full ¶

is_full() -> bool

Checks if the buffer is full.

Returns¶

bool: True if the buffer is full, False otherwise.

ring_buffer ¶

Ring Buffer¶

Alias for CircularBuffer providing ring buffer functionality.

Classes¶

sliding_window ¶

Sliding Window¶

A sliding window data structure for stream processing and moving calculations.

Classes¶

SlidingWindow ¶

SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)

SlidingWindow Class¶

A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.

Parameters¶

size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.

Initialize the sliding window.

size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.

Functions¶

add ¶

add(value: Any) -> Any | None

Add a value to the window and return aggregated result.

value: The value to add to the window.

Any | None: Aggregated result if aggregation_func is set.

clear ¶

clear() -> None

Clear all elements from window.

get_aggregate ¶

get_aggregate() -> Any | None

Get aggregated value of current window.

Any | None: Aggregated result or None if no aggregation function.

get_window ¶

get_window() -> list[Any]

Get current window contents.

list[Any]: Current window values in order.

is_empty ¶

is_empty() -> bool

Check if window is empty.

bool: True if window contains no elements.

is_full ¶

is_full() -> bool

Check if window is full.

bool: True if window has reached its size.

moving_average ¶

moving_average() -> float | None

Calculate moving average of numeric window.

float | None: Average of window values or None if empty.

moving_max ¶

moving_max() -> Any | None

Get maximum value in window.

Any | None: Maximum value or None if empty.

moving_min ¶

moving_min() -> Any | None

Get minimum value in window.

Any | None: Minimum value or None if empty.

moving_sum ¶

moving_sum() -> float

Calculate moving sum of numeric window.

float | None: Sum of window values or None if empty.

cache ¶

Cache Collections Module¶

Provides caching data structures with various eviction policies.

Classes:¶

LRUCache: Least Recently Used cache
LFUCache: Least Frequently Used cache
TTLCache: Time-To-Live cache

Classes¶

LFUCache ¶

LFUCache(capacity: int)

LFUCache Class¶

A Least Frequently Used (LFU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LFU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

LRUCache ¶

LRUCache(capacity: int)

LRUCache Class¶

A Least Recently Used (LRU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LRU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

items ¶

items() -> list[tuple[Any, Any]]

Get all key-value pairs.

list[tuple[Any, Any]]: List of (key, value) tuples.

keys ¶

keys() -> list[Any]

Get all keys in cache.

list[Any]: List of keys in order of use.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

values ¶

values() -> list[Any]

Get all values in cache.

list[Any]: List of values.

TTLCache ¶

TTLCache(default_ttl: float = 300.0, max_size: int | None = None)

TTLCache Class¶

A cache with Time-To-Live (TTL) expiration for entries.

Parameters¶

default_ttl : float Default time-to-live in seconds. max_size : int | None Optional maximum cache size.

Initialize a TTL cache.

default_ttl: Default expiration time in seconds.
max_size: Optional maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache if not expired.

key: The key to look up.

Any | None: The value or None if not found/expired.

get_ttl ¶

get_ttl(key: Any) -> float | None

Get remaining TTL for a key.

key: The key to check.

float | None: Remaining seconds or None if not found/expired.

put ¶

put(key: Any, value: Any, ttl: float | None = None) -> None

Add or update a key-value pair with TTL.

key: The key to store.
value: The value to store.
ttl: Optional TTL in seconds (uses default if not provided).

Modules¶

lfu_cache ¶

LFU Cache¶

Least Frequently Used (LFU) cache implementation.

Classes¶

LFUCache ¶

LFUCache(capacity: int)

LFUCache Class¶

A Least Frequently Used (LFU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LFU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

lru_cache ¶

LRU Cache¶

Least Recently Used (LRU) cache implementation.

Classes¶

LRUCache ¶

LRUCache(capacity: int)

LRUCache Class¶

A Least Recently Used (LRU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LRU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

items ¶

items() -> list[tuple[Any, Any]]

Get all key-value pairs.

list[tuple[Any, Any]]: List of (key, value) tuples.

keys ¶

keys() -> list[Any]

Get all keys in cache.

list[Any]: List of keys in order of use.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

values ¶

values() -> list[Any]

Get all values in cache.

list[Any]: List of values.

ttl_cache ¶

TTL Cache¶

Time-To-Live (TTL) cache with automatic expiration.

Classes¶

TTLCache ¶

TTLCache(default_ttl: float = 300.0, max_size: int | None = None)

TTLCache Class¶

A cache with Time-To-Live (TTL) expiration for entries.

Parameters¶

default_ttl : float Default time-to-live in seconds. max_size : int | None Optional maximum cache size.

Initialize a TTL cache.

default_ttl: Default expiration time in seconds.
max_size: Optional maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache if not expired.

key: The key to look up.

Any | None: The value or None if not found/expired.

get_ttl ¶

get_ttl(key: Any) -> float | None

Get remaining TTL for a key.

key: The key to check.

float | None: Remaining seconds or None if not found/expired.

put ¶

put(key: Any, value: Any, ttl: float | None = None) -> None

Add or update a key-value pair with TTL.

key: The key to store.
value: The value to store.
ttl: Optional TTL in seconds (uses default if not provided).

dict ¶

Dictionary Operations Module¶

Utilities for dictionary manipulation and transformation.

Functions¶

dict_merge: Merge multiple dictionaries.
dict_filter: Filter dictionary by predicate.
dict_invert: Swap keys and values.
dict_deep_get: Get nested value safely.
dict_deep_set: Set nested value, creating paths.

Examples¶

from rite.collections.dict import dict_merge, dict_filter dict_merge({"a": 1}, {"b": 2}) {'a': 1, 'b': 2} dict_filter({"a": 1, "b": 2}, lambda k, v: v > 1)

Modules¶

dict_deep_get ¶

Dict Deep Get¶

Safely get nested dictionary values using key paths.

Functions¶

dict_deep_get: Get nested value using list of keys.

Examples¶

from rite.collections.dict import dict_deep_get data = {"user": {"profile": {"name": "John"}}} dict_deep_get(data, ["user", "profile", "name"]) 'John'

Functions¶

dict_deep_get ¶

dict_deep_get(d: dict[str, Any], keys: list[str], default: Any = None) -> Any

Get nested dictionary value using key path.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to traverse.	required
`keys`	`list[str]`	List of keys forming path to desired value.	required
`default`	`Any`	Value to return if path doesn't exist.	`None`

Returns:

Type	Description
`Any`	Value at the key path or default.

Examples:

>>> data = {"user": {"profile": {"name": "John"}}}
>>> dict_deep_get(data, ["user", "profile", "name"])
'John'
>>> dict_deep_get(data, ["user", "missing"], "N/A")
'N/A'
>>> dict_deep_get({}, ["a", "b"])
None

dict_deep_set ¶

Dict Deep Set¶

Set nested dictionary values, creating intermediate dicts as needed.

Functions¶

dict_deep_set: Set value at nested key path.

Examples¶

from rite.collections.dict import dict_deep_set d = {} dict_deep_set(d, ["user", "profile", "name"], "John") d {'user': {'profile': {'name': 'John'}}}

Functions¶

dict_deep_set ¶

dict_deep_set(d: dict[str, Any], keys: list[str], value: Any) -> None

Set nested dictionary value, creating intermediate dicts.

Modifies dictionary in-place.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to modify.	required
`keys`	`list[str]`	List of keys forming path where value should be set.	required
`value`	`Any`	Value to set at the key path.	required

Examples:

>>> d = {}
>>> dict_deep_set(d, ["user", "profile", "name"], "John")
>>> d
{'user': {'profile': {'name': 'John'}}}
>>> dict_deep_set(d, ["user", "age"], 30)
>>> d['user']['age']
30

dict_filter ¶

Dict Filter¶

Filter dictionary entries based on key or value predicates.

Functions¶

dict_filter: Filter dictionary by predicate.

Examples¶

from rite.collections.dict import dict_filter dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)

Functions¶

dict_filter ¶

dict_filter(d: dict[K, V], predicate: Callable[[K, V], bool]) -> dict[K, V]

Filter dictionary entries based on predicate function.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to filter.	required
`predicate`	`Callable[[K, V], bool]`	Function taking (key, value) returning True to keep.	required

Returns:

Type	Description
`dict[K, V]`	New dictionary with filtered entries.

Examples:

>>> dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)
{'b': 2, 'c': 3}
>>> dict_filter({"a": 1, "b": 2}, lambda k, v: k == "a")
{'a': 1}
>>> dict_filter({}, lambda k, v: True)
{}

dict_invert ¶

Dict Invert¶

Invert dictionary keys and values.

Functions¶

dict_invert: Swap dictionary keys and values.

Examples¶

from rite.collections.dict import dict_invert dict_invert({"a": 1, "b": 2})

Functions¶

dict_invert ¶

dict_invert(d: dict[K, V]) -> dict[V, K]

Invert dictionary keys and values.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to invert.	required

Returns:

Type	Description
`dict[V, K]`	New dictionary with keys and values swapped.

Raises:

Type	Description
`TypeError`	If values are not hashable.

Examples:

>>> dict_invert({"a": 1, "b": 2})
{1: 'a', 2: 'b'}
>>> dict_invert({1: "one", 2: "two"})
{'one': 1, 'two': 2}
>>> dict_invert({})
{}

dict_merge ¶

Dict Merge¶

Merge multiple dictionaries with various strategies.

Functions¶

dict_merge: Merge dictionaries with configurable behavior.

Examples¶

from rite.collections.dict import dict_merge dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})

Functions¶

dict_merge ¶

dict_merge(*dicts: dict[K, V], deep: bool = False) -> dict[K, V]

Merge multiple dictionaries.

Later dictionaries override earlier ones for duplicate keys.

Parameters:

Name	Type	Description	Default
`*dicts`	`dict[K, V]`	Variable number of dictionaries to merge.	`()`
`deep`	`bool`	If True, recursively merge nested dicts.	`False`

Returns:

Type	Description
`dict[K, V]`	New merged dictionary.

Examples:

>>> dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})
{'a': 1, 'b': 3, 'c': 4}
>>> dict_merge({"a": 1}, {"b": 2}, {"c": 3})
{'a': 1, 'b': 2, 'c': 3}
>>> d1 = {"a": {"x": 1}}
>>> d2 = {"a": {"y": 2}}
>>> dict_merge(d1, d2, deep=True)
{'a': {'x': 1, 'y': 2}}

dict_deep_get ¶

Dict Deep Get¶

Safely get nested dictionary values using key paths.

Functions¶

dict_deep_get: Get nested value using list of keys.

Examples¶

from rite.collections.dict import dict_deep_get data = {"user": {"profile": {"name": "John"}}} dict_deep_get(data, ["user", "profile", "name"]) 'John'

Functions¶

dict_deep_get ¶

dict_deep_get(d: dict[str, Any], keys: list[str], default: Any = None) -> Any

Get nested dictionary value using key path.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to traverse.	required
`keys`	`list[str]`	List of keys forming path to desired value.	required
`default`	`Any`	Value to return if path doesn't exist.	`None`

Returns:

Type	Description
`Any`	Value at the key path or default.

Examples:

>>> data = {"user": {"profile": {"name": "John"}}}
>>> dict_deep_get(data, ["user", "profile", "name"])
'John'
>>> dict_deep_get(data, ["user", "missing"], "N/A")
'N/A'
>>> dict_deep_get({}, ["a", "b"])
None

dict_deep_set ¶

Dict Deep Set¶

Set nested dictionary values, creating intermediate dicts as needed.

Functions¶

dict_deep_set: Set value at nested key path.

Examples¶

from rite.collections.dict import dict_deep_set d = {} dict_deep_set(d, ["user", "profile", "name"], "John") d {'user': {'profile': {'name': 'John'}}}

Functions¶

dict_deep_set ¶

dict_deep_set(d: dict[str, Any], keys: list[str], value: Any) -> None

Set nested dictionary value, creating intermediate dicts.

Modifies dictionary in-place.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to modify.	required
`keys`	`list[str]`	List of keys forming path where value should be set.	required
`value`	`Any`	Value to set at the key path.	required

Examples:

>>> d = {}
>>> dict_deep_set(d, ["user", "profile", "name"], "John")
>>> d
{'user': {'profile': {'name': 'John'}}}
>>> dict_deep_set(d, ["user", "age"], 30)
>>> d['user']['age']
30

dict_filter ¶

Dict Filter¶

Filter dictionary entries based on key or value predicates.

Functions¶

dict_filter: Filter dictionary by predicate.

Examples¶

from rite.collections.dict import dict_filter dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)

Functions¶

dict_filter ¶

dict_filter(d: dict[K, V], predicate: Callable[[K, V], bool]) -> dict[K, V]

Filter dictionary entries based on predicate function.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to filter.	required
`predicate`	`Callable[[K, V], bool]`	Function taking (key, value) returning True to keep.	required

Returns:

Type	Description
`dict[K, V]`	New dictionary with filtered entries.

Examples:

>>> dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)
{'b': 2, 'c': 3}
>>> dict_filter({"a": 1, "b": 2}, lambda k, v: k == "a")
{'a': 1}
>>> dict_filter({}, lambda k, v: True)
{}

dict_invert ¶

Dict Invert¶

Invert dictionary keys and values.

Functions¶

dict_invert: Swap dictionary keys and values.

Examples¶

from rite.collections.dict import dict_invert dict_invert({"a": 1, "b": 2})

Functions¶

dict_invert ¶

dict_invert(d: dict[K, V]) -> dict[V, K]

Invert dictionary keys and values.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to invert.	required

Returns:

Type	Description
`dict[V, K]`	New dictionary with keys and values swapped.

Raises:

Type	Description
`TypeError`	If values are not hashable.

Examples:

>>> dict_invert({"a": 1, "b": 2})
{1: 'a', 2: 'b'}
>>> dict_invert({1: "one", 2: "two"})
{'one': 1, 'two': 2}
>>> dict_invert({})
{}

dict_merge ¶

Dict Merge¶

Merge multiple dictionaries with various strategies.

Functions¶

dict_merge: Merge dictionaries with configurable behavior.

Examples¶

from rite.collections.dict import dict_merge dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})

Functions¶

dict_merge ¶

dict_merge(*dicts: dict[K, V], deep: bool = False) -> dict[K, V]

Merge multiple dictionaries.

Later dictionaries override earlier ones for duplicate keys.

Parameters:

Name	Type	Description	Default
`*dicts`	`dict[K, V]`	Variable number of dictionaries to merge.	`()`
`deep`	`bool`	If True, recursively merge nested dicts.	`False`

Returns:

Type	Description
`dict[K, V]`	New merged dictionary.

Examples:

>>> dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})
{'a': 1, 'b': 3, 'c': 4}
>>> dict_merge({"a": 1}, {"b": 2}, {"c": 3})
{'a': 1, 'b': 2, 'c': 3}
>>> d1 = {"a": {"x": 1}}
>>> d2 = {"a": {"y": 2}}
>>> dict_merge(d1, d2, deep=True)
{'a': {'x': 1, 'y': 2}}

list ¶

List Operations Module¶

Utilities for list manipulation and transformation.

Functions¶

list_unique: Remove duplicates while preserving order.
list_flatten: Flatten nested lists recursively.
list_chunk: Split list into chunks.
list_group_by: Group items by key function or attribute.
list_partition: Split list into two based on predicate.
list_interleave: Interleave multiple lists.

Examples¶

from rite.collections.list import list_unique, list_chunk list_unique([1, 2, 2, 3]) [1, 2, 3] list_chunk([1, 2, 3, 4, 5], size=2) [[1, 2], [3, 4], [5]]

Modules¶

list_chunk ¶

List Chunk¶

Split a list into chunks of specified size.

Functions¶

list_chunk: Split list into smaller chunks.

Examples¶

from rite.collections.list import list_chunk list_chunk([1, 2, 3, 4, 5], size=2) [[1, 2], [3, 4], [5]] list_chunk([1, 2, 3, 4, 5, 6], size=3) [[1, 2, 3], [4, 5, 6]]

Functions¶

list_chunk ¶

list_chunk(items: list[T], size: int) -> list[list[T]]

Split a list into chunks of specified size.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to split into chunks.	required
`size`	`int`	Size of each chunk. Must be positive.	required

Returns:

Type	Description
`list[list[T]]`	List of chunks, last chunk may be smaller.

Raises:

Type	Description
`ValueError`	If size is not positive.

Examples:

>>> list_chunk([1, 2, 3, 4, 5], size=2)
[[1, 2], [3, 4], [5]]
>>> list_chunk([1, 2, 3, 4, 5, 6], size=3)
[[1, 2, 3], [4, 5, 6]]
>>> list_chunk([], size=2)
[]
>>> list_chunk([1, 2, 3], size=5)
[[1, 2, 3]]

list_flatten ¶

List Flatten¶

Flatten nested lists into a single-level list.

Functions¶

list_flatten: Flatten nested lists recursively.

Examples¶

from rite.collections.list import list_flatten list_flatten([[1, 2], [3, 4], [5]]) [1, 2, 3, 4, 5] list_flatten([[1, [2, 3]], [4, 5]]) [1, 2, 3, 4, 5]

Functions¶

list_flatten ¶

list_flatten(items: list[Any], depth: int | None = None) -> list[Any]

Flatten nested lists recursively.

Parameters:

Name	Type	Description	Default
`items`	`list[Any]`	List that may contain nested lists.	required
`depth`	`int \| None`	Maximum depth to flatten. None for full recursion.	`None`

Returns:

Type	Description
`list[Any]`	Flattened list.

Examples:

>>> list_flatten([[1, 2], [3, 4], [5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, 3]], [4, 5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, [3]]], [4]], depth=1)
[1, [2, [3]], 4]
>>> list_flatten([])
[]

list_group_by ¶

List Group By¶

Group list items by a key function or attribute.

Functions¶

list_group_by: Group items by key.

Examples¶

from rite.collections.list import list_group_by data = [{"type": "fruit", "name": "apple"}] list_group_by(data, key="type") {'fruit': [{'type': 'fruit', 'name': 'apple'}]}

Functions¶

list_group_by ¶

list_group_by(items: list[T], key: str | Callable[[T], K]) -> dict[Any, list[T]]

Group list items by a key function or dict key.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List of items to group.	required
`key`	`str \| Callable[[T], K]`	Either attribute name (str) or function to extract key.	required

Returns:

Type	Description
`dict[Any, list[T]]`	Dictionary mapping keys to lists of items.

Examples:

>>> data = [
...     {"type": "fruit", "name": "apple"},
...     {"type": "fruit", "name": "banana"},
...     {"type": "veg", "name": "carrot"}
... ]
>>> list_group_by(data, key="type")
{'fruit': [...], 'veg': [...]}
>>> list_group_by([1, 2, 3, 4], key=lambda x: x % 2)
{1: [1, 3], 0: [2, 4]}

list_interleave ¶

List Interleave¶

Interleave multiple lists together.

Functions¶

list_interleave: Combine lists by alternating elements.

Examples¶

from rite.collections.list import list_interleave list_interleave([1, 2, 3], ['a', 'b', 'c']) [1, 'a', 2, 'b', 3, 'c']

Functions¶

list_interleave ¶

list_interleave(*lists: list[Any]) -> list[Any]

Interleave multiple lists by alternating elements.

Parameters:

Name	Type	Description	Default
`*lists`	`list[Any]`	Variable number of lists to interleave.	`()`

Returns:

Type	Description
`list[Any]`	New list with elements from all lists interleaved.

Examples:

>>> list_interleave([1, 2, 3], ['a', 'b', 'c'])
[1, 'a', 2, 'b', 3, 'c']
>>> list_interleave([1, 2], [10, 20], [100, 200])
[1, 10, 100, 2, 20, 200]
>>> list_interleave([1, 2, 3], ['a'])
[1, 'a', 2, 3]
>>> list_interleave([])
[]

list_partition ¶

List Partition¶

Partition a list into two lists based on a predicate function.

Functions¶

list_partition: Split list into matching and non-matching items.

Examples¶

from rite.collections.list import list_partition list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0) ([2, 4], [1, 3, 5])

Functions¶

list_partition ¶

list_partition(items: list[T], predicate: Callable[[T], bool]) -> tuple[list[T], list[T]]

Partition a list based on a predicate function.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to partition.	required
`predicate`	`Callable[[T], bool]`	Function that returns True for items in first list.	required

Returns:

Type	Description
`tuple[list[T], list[T]]`	Tuple of (matching_items, non_matching_items).

Examples:

>>> list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0)
([2, 4], [1, 3, 5])
>>> list_partition(['a', 'bb', 'ccc'], lambda x: len(x) > 1)
(['bb', 'ccc'], ['a'])
>>> list_partition([], lambda x: True)
([], [])

list_unique ¶

List Unique¶

Remove duplicate elements from a list while preserving order.

Functions¶

list_unique: Remove duplicates from a list.

Examples¶

from rite.collections.list import list_unique list_unique([1, 2, 2, 3, 3, 3]) [1, 2, 3] list_unique(['a', 'b', 'a', 'c']) ['a', 'b', 'c']

Functions¶

list_unique ¶

list_unique(items: list[T]) -> list[T]

Remove duplicate elements while preserving order.

Uses a set to track seen elements and preserves first occurrence of each unique element.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List that may contain duplicates.	required

Returns:

Type	Description
`list[T]`	New list with duplicates removed, order preserved.

Examples:

>>> list_unique([1, 2, 2, 3, 3, 3])
[1, 2, 3]
>>> list_unique(['a', 'b', 'a', 'c'])
['a', 'b', 'c']
>>> list_unique([])
[]

list_chunk ¶

List Chunk¶

Split a list into chunks of specified size.

Functions¶

list_chunk: Split list into smaller chunks.

Examples¶

from rite.collections.list import list_chunk list_chunk([1, 2, 3, 4, 5], size=2) [[1, 2], [3, 4], [5]] list_chunk([1, 2, 3, 4, 5, 6], size=3) [[1, 2, 3], [4, 5, 6]]

Functions¶

list_chunk ¶

list_chunk(items: list[T], size: int) -> list[list[T]]

Split a list into chunks of specified size.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to split into chunks.	required
`size`	`int`	Size of each chunk. Must be positive.	required

Returns:

Type	Description
`list[list[T]]`	List of chunks, last chunk may be smaller.

Raises:

Type	Description
`ValueError`	If size is not positive.

Examples:

>>> list_chunk([1, 2, 3, 4, 5], size=2)
[[1, 2], [3, 4], [5]]
>>> list_chunk([1, 2, 3, 4, 5, 6], size=3)
[[1, 2, 3], [4, 5, 6]]
>>> list_chunk([], size=2)
[]
>>> list_chunk([1, 2, 3], size=5)
[[1, 2, 3]]

list_flatten ¶

List Flatten¶

Flatten nested lists into a single-level list.

Functions¶

list_flatten: Flatten nested lists recursively.

Examples¶

from rite.collections.list import list_flatten list_flatten([[1, 2], [3, 4], [5]]) [1, 2, 3, 4, 5] list_flatten([[1, [2, 3]], [4, 5]]) [1, 2, 3, 4, 5]

Functions¶

list_flatten ¶

list_flatten(items: list[Any], depth: int | None = None) -> list[Any]

Flatten nested lists recursively.

Parameters:

Name	Type	Description	Default
`items`	`list[Any]`	List that may contain nested lists.	required
`depth`	`int \| None`	Maximum depth to flatten. None for full recursion.	`None`

Returns:

Type	Description
`list[Any]`	Flattened list.

Examples:

>>> list_flatten([[1, 2], [3, 4], [5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, 3]], [4, 5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, [3]]], [4]], depth=1)
[1, [2, [3]], 4]
>>> list_flatten([])
[]

list_group_by ¶

List Group By¶

Group list items by a key function or attribute.

Functions¶

list_group_by: Group items by key.

Examples¶

from rite.collections.list import list_group_by data = [{"type": "fruit", "name": "apple"}] list_group_by(data, key="type") {'fruit': [{'type': 'fruit', 'name': 'apple'}]}

Functions¶

list_group_by ¶

list_group_by(items: list[T], key: str | Callable[[T], K]) -> dict[Any, list[T]]

Group list items by a key function or dict key.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List of items to group.	required
`key`	`str \| Callable[[T], K]`	Either attribute name (str) or function to extract key.	required

Returns:

Type	Description
`dict[Any, list[T]]`	Dictionary mapping keys to lists of items.

Examples:

>>> data = [
...     {"type": "fruit", "name": "apple"},
...     {"type": "fruit", "name": "banana"},
...     {"type": "veg", "name": "carrot"}
... ]
>>> list_group_by(data, key="type")
{'fruit': [...], 'veg': [...]}
>>> list_group_by([1, 2, 3, 4], key=lambda x: x % 2)
{1: [1, 3], 0: [2, 4]}

list_interleave ¶

List Interleave¶

Interleave multiple lists together.

Functions¶

list_interleave: Combine lists by alternating elements.

Examples¶

from rite.collections.list import list_interleave list_interleave([1, 2, 3], ['a', 'b', 'c']) [1, 'a', 2, 'b', 3, 'c']

Functions¶

list_interleave ¶

list_interleave(*lists: list[Any]) -> list[Any]

Interleave multiple lists by alternating elements.

Parameters:

Name	Type	Description	Default
`*lists`	`list[Any]`	Variable number of lists to interleave.	`()`

Returns:

Type	Description
`list[Any]`	New list with elements from all lists interleaved.

Examples:

>>> list_interleave([1, 2, 3], ['a', 'b', 'c'])
[1, 'a', 2, 'b', 3, 'c']
>>> list_interleave([1, 2], [10, 20], [100, 200])
[1, 10, 100, 2, 20, 200]
>>> list_interleave([1, 2, 3], ['a'])
[1, 'a', 2, 3]
>>> list_interleave([])
[]

list_partition ¶

List Partition¶

Partition a list into two lists based on a predicate function.

Functions¶

list_partition: Split list into matching and non-matching items.

Examples¶

from rite.collections.list import list_partition list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0) ([2, 4], [1, 3, 5])

Functions¶

list_partition ¶

list_partition(items: list[T], predicate: Callable[[T], bool]) -> tuple[list[T], list[T]]

Partition a list based on a predicate function.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to partition.	required
`predicate`	`Callable[[T], bool]`	Function that returns True for items in first list.	required

Returns:

Type	Description
`tuple[list[T], list[T]]`	Tuple of (matching_items, non_matching_items).

Examples:

>>> list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0)
([2, 4], [1, 3, 5])
>>> list_partition(['a', 'bb', 'ccc'], lambda x: len(x) > 1)
(['bb', 'ccc'], ['a'])
>>> list_partition([], lambda x: True)
([], [])

list_unique ¶

List Unique¶

Remove duplicate elements from a list while preserving order.

Functions¶

list_unique: Remove duplicates from a list.

Examples¶

from rite.collections.list import list_unique list_unique([1, 2, 2, 3, 3, 3]) [1, 2, 3] list_unique(['a', 'b', 'a', 'c']) ['a', 'b', 'c']

Functions¶

list_unique ¶

list_unique(items: list[T]) -> list[T]

Remove duplicate elements while preserving order.

Uses a set to track seen elements and preserves first occurrence of each unique element.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List that may contain duplicates.	required

Returns:

Type	Description
`list[T]`	New list with duplicates removed, order preserved.

Examples:

>>> list_unique([1, 2, 2, 3, 3, 3])
[1, 2, 3]
>>> list_unique(['a', 'b', 'a', 'c'])
['a', 'b', 'c']
>>> list_unique([])
[]

pattern ¶

Design Pattern Collections Module¶

Provides design pattern implementations for common software patterns.

Classes:¶

SingletonMeta: Singleton pattern metaclass
Observer: Observer pattern implementation
ObjectPool: Object pool pattern

Classes¶

ObjectPool ¶

ObjectPool(factory: Callable[[], Any], max_size: int = 10, reset: Callable[[Any], None] | None = None)

ObjectPool Class¶

A generic object pool for managing reusable objects.

Parameters¶

factory : Callable[[], Any] Factory function to create new objects. max_size : int Maximum number of objects in the pool. reset_func : Callable[[Any], None] | None Optional function to reset objects before reuse.

Initialize an object pool.

factory: Function to create new objects.
max_size: Maximum pool size.
reset_func: Optional function to reset objects.

Functions¶

acquire ¶

acquire() -> Any

Acquire an object from the pool.

Any: An object from the pool or newly created.

available_count ¶

available_count() -> int

Get number of available objects.

int: Number of available objects.

clear ¶

clear() -> None

Clear all available objects from the pool.

in_use_count ¶

in_use_count() -> int

Get number of objects currently in use.

int: Number of in-use objects.

release ¶

release(obj: Any) -> None

Release an object back to the pool.

obj: The object to release.

ValueError: If object was not acquired from this pool.

size ¶

size() -> int

Get total number of objects managed by pool.

int: Total objects (available + in use).

Observable ¶

Observable()

Observable Class¶

Base class for observable objects in the Observer pattern.

Initialize an observable object.

Attributes¶

observers property ¶

observers: tuple[Observer, ...]

Return observers as an immutable tuple.

Functions¶

attach ¶

attach(observer: Observer) -> None

Attach an observer.

observer: The observer to attach.

clear_observers ¶

clear_observers() -> None

Remove all observers.

detach ¶

detach(observer: Observer) -> None

Detach an observer.

observer: The observer to detach.

get_observer_count ¶

get_observer_count() -> int

Get the number of attached observers.

int: Number of observers.

notify ¶

notify(*args: Any, **kwargs: Any) -> None

Notify all observers of a change.

*args: Positional arguments to pass to observers.
**kwargs: Keyword arguments to pass to observers.

Observer ¶

Observer Abstract Base Class¶

Base class for observers in the Observer pattern.

Functions¶

update ¶

update(observable: Observable | None, *args: Any, **kwargs: Any) -> None

Called when the observed object changes.

SingletonMeta ¶

Bases: type

SingletonMeta Class¶

A metaclass for implementing the Singleton pattern. This ensures that only one instance of a class exists throughout the application lifecycle.

Attributes¶

_instances : dict[type, Any] A dictionary that maps classes to their single instances.

Methods¶

call(*args, **kwargs): Overrides the __call__ method to control instance creation.

Example¶

# Define a Singleton class
class Configuration(metaclass=SingletonMeta):
    def __init__(self, value):
        self.value = value

Functions¶

reset_instance classmethod ¶

reset_instance(target_cls: type) -> None

Remove a cached instance for the given class if present.

Modules¶

object_pool ¶

Object Pool Pattern¶

Implementation of the Object Pool design pattern for resource management.

Classes¶

ObjectPool ¶

ObjectPool(factory: Callable[[], Any], max_size: int = 10, reset: Callable[[Any], None] | None = None)

ObjectPool Class¶

A generic object pool for managing reusable objects.

Parameters¶

factory : Callable[[], Any] Factory function to create new objects. max_size : int Maximum number of objects in the pool. reset_func : Callable[[Any], None] | None Optional function to reset objects before reuse.

Initialize an object pool.

factory: Function to create new objects.
max_size: Maximum pool size.
reset_func: Optional function to reset objects.

Functions¶

acquire ¶

acquire() -> Any

Acquire an object from the pool.

Any: An object from the pool or newly created.

available_count ¶

available_count() -> int

Get number of available objects.

int: Number of available objects.

clear ¶

clear() -> None

Clear all available objects from the pool.

in_use_count ¶

in_use_count() -> int

Get number of objects currently in use.

int: Number of in-use objects.

release ¶

release(obj: Any) -> None

Release an object back to the pool.

obj: The object to release.

ValueError: If object was not acquired from this pool.

size ¶

size() -> int

Get total number of objects managed by pool.

int: Total objects (available + in use).

observer ¶

Observer Pattern¶

Implementation of the Observer design pattern for event-driven programming.

Classes¶

Observable ¶

Observable()

Observable Class¶

Base class for observable objects in the Observer pattern.

Initialize an observable object.

Attributes¶

observers property ¶

observers: tuple[Observer, ...]

Return observers as an immutable tuple.

Functions¶

attach ¶

attach(observer: Observer) -> None

Attach an observer.

observer: The observer to attach.

clear_observers ¶

clear_observers() -> None

Remove all observers.

detach ¶

detach(observer: Observer) -> None

Detach an observer.

observer: The observer to detach.

get_observer_count ¶

get_observer_count() -> int

Get the number of attached observers.

int: Number of observers.

notify ¶

notify(*args: Any, **kwargs: Any) -> None

Notify all observers of a change.

*args: Positional arguments to pass to observers.
**kwargs: Keyword arguments to pass to observers.

Observer ¶

Observer Abstract Base Class¶

Base class for observers in the Observer pattern.

Functions¶

update ¶

update(observable: Observable | None, *args: Any, **kwargs: Any) -> None

Called when the observed object changes.

singleton ¶

Singleton Module¶

This module provides a metaclass SingletonMeta for implementing the Singleton design pattern. Classes using this metaclass ensure that only one instance of the class is created, regardless of how many times it is instantiated.

Classes:¶

SingletonMeta: A metaclass for implementing the Singleton pattern.

Features:¶

Ensures a single instance of a class.
Thread-safe by default.
Provides an easy-to-use interface for creating Singleton classes.

Usage:¶

To use the Singleton pattern, define your class with SingletonMeta as its metaclass:

class MySingleton(metaclass=SingletonMeta):
    pass  # pylint: disable=unnecessary-pass

Example:¶

# Define a Singleton class
class Configuration(metaclass=SingletonMeta):
    def __init__(self, value):
        self.value = value

# Test the Singleton behavior
config1 = Configuration("value1")
config2 = Configuration("value2")
assert config1 is config2  # True
print(config1.value)       # Outputs: value2

Classes¶

Configuration ¶

Configuration(value: str)

A Singleton class for storing configuration values.

SingletonMeta ¶

Bases: type

SingletonMeta Class¶

A metaclass for implementing the Singleton pattern. This ensures that only one instance of a class exists throughout the application lifecycle.

Attributes¶

_instances : dict[type, Any] A dictionary that maps classes to their single instances.

Methods¶

call(*args, **kwargs): Overrides the __call__ method to control instance creation.

Example¶

# Define a Singleton class
class Configuration(metaclass=SingletonMeta):
    def __init__(self, value):
        self.value = value

Functions¶

reset_instance classmethod ¶

reset_instance(target_cls: type) -> None

Remove a cached instance for the given class if present.

queue ¶

Queue Collections Module¶

Provides specialized queue implementations.

Classes:¶

PriorityQueue: Priority-based queue
DequeWrapper: Enhanced deque wrapper
CircularQueue: Circular queue implementation

Classes¶

CircularQueue ¶

CircularQueue(capacity: int)

CircularQueue Class¶

A fixed-size circular queue (FIFO).

Initialize a circular queue.

capacity: Maximum capacity of the queue.

Functions¶

clear ¶

clear() -> None

Clear all items from queue.

dequeue ¶

dequeue() -> Any

Remove and return item from front of queue.

Any: The front item.

enqueue ¶

enqueue(item: Any) -> bool

Add an item to the rear of the queue.

item: Item to add.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

is_full ¶

is_full() -> bool

Check if queue is full.

peek ¶

peek() -> Any

Return front item without removing it.

Any: The front item.

DequeWrapper ¶

DequeWrapper(max_size: int | None = None)

DequeWrapper Class¶

A wrapper around collections.deque with additional utilities.

Initialize a deque wrapper.

max_size: Maximum length of the deque.

Functions¶

clear ¶

clear() -> None

Remove all items.

is_empty ¶

is_empty() -> bool

Check if deque is empty.

peek_left ¶

peek_left() -> Any

Return item at left end without removing.

peek_right ¶

peek_right() -> Any

Return item at right end without removing.

pop_left ¶

pop_left() -> Any

Remove and return item from left end.

pop_right ¶

pop_right() -> Any

Remove and return item from right end.

push_left ¶

push_left(item: Any) -> None

Add item to the left end.

push_right ¶

push_right(item: Any) -> None

Add item to the right end.

rotate ¶

rotate(n: int = 1) -> None

Rotate deque n steps to the right.

n: Number of steps to rotate (negative for left rotation).

to_list ¶

to_list() -> list[Any]

Convert to list.

PriorityQueue ¶

PriorityQueue()

PriorityQueue Class¶

A priority queue where items with lower priority values are dequeued first.

Initialize an empty priority queue.

Functions¶

clear ¶

clear() -> None

Remove all items from the queue.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

bool: True if queue is empty.

peek ¶

peek() -> Any

Return the highest priority item without removing it.

Any: The highest priority item.

pop ¶

pop() -> Any

Remove and return the highest priority item.

Any: The highest priority item.

push ¶

push(item: Any, priority: float = 0.0) -> None

Add an item with a given priority.

item: The item to add.
priority: Priority value (lower = higher priority).

Modules¶

circular_queue ¶

Circular Queue¶

A circular queue with fixed capacity.

Classes¶

CircularQueue ¶

CircularQueue(capacity: int)

CircularQueue Class¶

A fixed-size circular queue (FIFO).

Initialize a circular queue.

capacity: Maximum capacity of the queue.

Functions¶

clear ¶

clear() -> None

Clear all items from queue.

dequeue ¶

dequeue() -> Any

Remove and return item from front of queue.

Any: The front item.

enqueue ¶

enqueue(item: Any) -> bool

Add an item to the rear of the queue.

item: Item to add.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

is_full ¶

is_full() -> bool

Check if queue is full.

peek ¶

peek() -> Any

Return front item without removing it.

Any: The front item.

deque_wrapper ¶

Deque Wrapper¶

Enhanced deque wrapper with additional functionality.

Classes¶

DequeWrapper ¶

DequeWrapper(max_size: int | None = None)

DequeWrapper Class¶

A wrapper around collections.deque with additional utilities.

Initialize a deque wrapper.

max_size: Maximum length of the deque.

Functions¶

clear ¶

clear() -> None

Remove all items.

is_empty ¶

is_empty() -> bool

Check if deque is empty.

peek_left ¶

peek_left() -> Any

Return item at left end without removing.

peek_right ¶

peek_right() -> Any

Return item at right end without removing.

pop_left ¶

pop_left() -> Any

Remove and return item from left end.

pop_right ¶

pop_right() -> Any

Remove and return item from right end.

push_left ¶

push_left(item: Any) -> None

Add item to the left end.

push_right ¶

push_right(item: Any) -> None

Add item to the right end.

rotate ¶

rotate(n: int = 1) -> None

Rotate deque n steps to the right.

n: Number of steps to rotate (negative for left rotation).

to_list ¶

to_list() -> list[Any]

Convert to list.

priority_queue ¶

Priority Queue¶

A priority queue implementation using heapq.

Classes¶

PriorityQueue ¶

PriorityQueue()

PriorityQueue Class¶

A priority queue where items with lower priority values are dequeued first.

Initialize an empty priority queue.

Functions¶

clear ¶

clear() -> None

Remove all items from the queue.

is_empty ¶

is_empty() -> bool

Check if queue is empty.

bool: True if queue is empty.

peek ¶

peek() -> Any

Return the highest priority item without removing it.

Any: The highest priority item.

pop ¶

pop() -> Any

Remove and return the highest priority item.

Any: The highest priority item.

push ¶

push(item: Any, priority: float = 0.0) -> None

Add an item with a given priority.

item: The item to add.
priority: Priority value (lower = higher priority).

set ¶

Set Operations Module¶

Utilities for set operations and transformations.

Functions¶

set_union: Union of multiple sets.
set_intersection: Intersection of multiple sets.
set_difference: Difference of sets.
set_symmetric_difference: Symmetric difference of two sets.

Examples¶

from rite.collections.set import set_union set_union({1, 2}, {2, 3}, {3, 4})

Functions¶

set_difference ¶

set_difference(first: set[T], *others: set[T]) -> set[T]

Return difference of first set minus all others.

Parameters:

Name	Type	Description	Default
`first`	`set[T]`	Base set to subtract from.	required
`*others`	`set[T]`	Sets to subtract.	`()`

Returns:

Type	Description
`set[T]`	New set with elements in first but not in others.

Examples:

>>> set_difference({1, 2, 3}, {2}, {3})
{1}
>>> set_difference({1, 2, 3, 4}, {2, 3})
{1, 4}

set_intersection ¶

set_intersection(*sets: set[T]) -> set[T]

Return intersection of multiple sets.

Parameters:

Name	Type	Description	Default
`*sets`	`set[T]`	Variable number of sets to intersect.	`()`

Returns:

Type	Description
`set[T]`	New set containing only common elements.

Examples:

>>> set_intersection({1, 2, 3}, {2, 3, 4}, {2, 3, 5})
{2, 3}
>>> set_intersection({1, 2}, {3, 4})
set()

set_symmetric_difference ¶

set_symmetric_difference(first: set[T], second: set[T]) -> set[T]

Return symmetric difference (elements in either but not both).

Parameters:

Name	Type	Description	Default
`first`	`set[T]`	First set.	required
`second`	`set[T]`	Second set.	required

Returns:

Type	Description
`set[T]`	New set with elements in first or second but not both.

Examples:

>>> set_symmetric_difference({1, 2, 3}, {2, 3, 4})
{1, 4}
>>> set_symmetric_difference({1, 2}, {3, 4})
{1, 2, 3, 4}

set_union ¶

set_union(*sets: set[T]) -> set[T]

Return union of multiple sets.

Parameters:

Name	Type	Description	Default
`*sets`	`set[T]`	Variable number of sets to unite.	`()`

Returns:

Type	Description
`set[T]`	New set containing all unique elements.

Examples:

>>> set_union({1, 2}, {2, 3}, {3, 4})
{1, 2, 3, 4}
>>> set_union({1}, {2}, {3})
{1, 2, 3}

tree ¶

Tree Collections Module¶

Provides tree data structures and hierarchical data management.

Classes:¶

NestedSetStructure: Hierarchical parent-child relationship manager
TreeNode: Generic tree node with children
BinaryTreeNode: Binary tree node implementation
Trie: Prefix tree for string operations

Classes¶

BinaryTreeNode ¶

BinaryTreeNode(value: Any, left: BinaryTreeNode | None = None, right: BinaryTreeNode | None = None)

BinaryTreeNode Class¶

A binary tree node with left and right children.

Parameters¶

value : Any The value stored in this node. left : BinaryTreeNode | None Left child node. right : BinaryTreeNode | None Right child node.

Initialize a binary tree node.

value: The value to store.
left: Left child node.
right: Right child node.

Functions¶

get_height ¶

get_height() -> int

Get height of subtree rooted at this node.

int: Height of subtree.

has_left_child ¶

has_left_child() -> bool

Check if node has left child.

bool: True if left child exists.

has_right_child ¶

has_right_child() -> bool

Check if node has right child.

bool: True if right child exists.

inorder_traversal ¶

inorder_traversal() -> list[BinaryTreeNode]

Perform in-order traversal (left, root, right).

list[Any]: Values in in-order.

is_leaf ¶

is_leaf() -> bool

Check if node is a leaf.

bool: True if node has no children.

postorder_traversal ¶

postorder_traversal() -> list[BinaryTreeNode]

Perform post-order traversal (left, right, root).

list[Any]: Values in post-order.

preorder_traversal ¶

preorder_traversal() -> list[BinaryTreeNode]

Perform pre-order traversal (root, left, right).

list[Any]: Values in pre-order.

traverse_inorder ¶

traverse_inorder() -> list[BinaryTreeNode]

Alias for inorder_traversal for compatibility.

traverse_postorder ¶

traverse_postorder() -> list[BinaryTreeNode]

Alias for postorder_traversal for compatibility.

traverse_preorder ¶

traverse_preorder() -> list[BinaryTreeNode]

Alias for preorder_traversal for compatibility.

NestedSetStructure ¶

NestedSetStructure()

Nested Set Structure Class¶

A structure to track items and their hierarchical relationships. Supports multi-level trees with parent/child tracking. Useful for scenarios like recursive publishing, dependency resolution, or graph-like traversal.

Initialize the nested set structure.

Functions¶

add ¶

add(item: Any, parent: Any | None = None) -> None

Add an item to the structure, optionally specifying a parent.

children ¶

children(item: Any) -> list[Any]

Return the children of the given item. If the item has no children, returns an empty list.

nested_items ¶

nested_items() -> list[Any]

Return a nested, flattened list of items (including children) in insertion order. Maintains hierarchy order.

original ¶

original(item: Any) -> Any

Return the original object stored in the set that equals item, or item itself if not found. This is useful for matching back to canonical instances.

parent ¶

parent(item: Any) -> Any | None

Return the parent of the given item, or None if it has no parent.

TreeNode ¶

TreeNode(value: Any, children: list[Self] | None = None)

TreeNode Class¶

A generic tree node that can have multiple children.

Parameters¶

value : Any The value stored in this node. children : list[TreeNode] | None Optional list of child nodes.

Initialize a tree node.

value: The value to store in this node.
children: Optional list of child nodes.

Functions¶

add_child ¶

add_child(child: Self) -> None

Add a child node.

child: The child node to add.

get_ancestors ¶

get_ancestors() -> list[Self]

Get all ancestor nodes from parent to root.

list[TreeNode]: List of ancestors in order (parent to root).

get_depth ¶

get_depth() -> int

Get the depth of this node (distance from root).

int: Depth of node (root is 0).

get_height ¶

get_height() -> int

Get the height of the subtree rooted at this node.

int: Height of subtree (leaf is 0).

get_siblings ¶

get_siblings() -> list[Self]

Get all sibling nodes (nodes with same parent).

list[TreeNode]: List of sibling nodes.

is_leaf ¶

is_leaf() -> bool

Check if this node is a leaf (has no children).

bool: True if node has no children.

is_root ¶

is_root() -> bool

Check if this node is a root (has no parent).

bool: True if node has no parent.

remove_child ¶

remove_child(child: Self) -> bool

Remove a child node.

child: The child node to remove.

bool: True if child was removed, False if not found.

traverse_levelorder ¶

traverse_levelorder() -> list[Self]

Traverse tree in level-order (breadth-first).

list[TreeNode]: Nodes in level-order.

traverse_postorder ¶

traverse_postorder() -> list[Self]

Traverse tree in post-order (children, then root).

list[TreeNode]: Nodes in post-order.

traverse_preorder ¶

traverse_preorder() -> list[Self]

Traverse tree in pre-order (root, then children).

list[TreeNode]: Nodes in pre-order.

Trie ¶

Trie()

Trie Class¶

A prefix tree (trie) for efficient string operations like autocomplete, spell checking, and prefix matching.

Initialize an empty trie.

Functions¶

autocomplete ¶

autocomplete(prefix: str) -> list[str]

Return all completions for the given prefix.

delete ¶

delete(word: str) -> bool

Delete a word from the trie.

word: The word to delete.

bool: True if word was deleted, False if not found.

get ¶

get(word: str) -> Any | None

Get the value associated with a word.

word: The word to look up.

Any | None: Associated value or None if not found.

get_all_words ¶

get_all_words() -> list[str]

Get all words stored in the trie.

list[str]: All words in the trie.

get_words_with_prefix ¶

get_words_with_prefix(prefix: str) -> list[str]

Get all words that start with given prefix.

prefix: The prefix to match.

list[str]: List of words with the prefix.

insert ¶

insert(word: str, value: Any | None = None) -> None

Insert a word into the trie.

word: The word to insert.
value: Optional value to associate with the word.

search ¶

search(word: str) -> bool

Search for an exact word in the trie.

word: The word to search for.

bool: True if word exists in trie.

starts_with ¶

starts_with(prefix: str) -> bool

Check if any word in trie starts with given prefix.

prefix: The prefix to check.

bool: True if prefix exists in trie.

Modules¶

binary_tree_node ¶

Binary Tree Node¶

Binary tree node with left and right children.

Classes¶

BinaryTreeNode ¶

BinaryTreeNode(value: Any, left: BinaryTreeNode | None = None, right: BinaryTreeNode | None = None)

BinaryTreeNode Class¶

A binary tree node with left and right children.

Parameters¶

value : Any The value stored in this node. left : BinaryTreeNode | None Left child node. right : BinaryTreeNode | None Right child node.

Initialize a binary tree node.

value: The value to store.
left: Left child node.
right: Right child node.

Functions¶

get_height ¶

get_height() -> int

Get height of subtree rooted at this node.

int: Height of subtree.

has_left_child ¶

has_left_child() -> bool

Check if node has left child.

bool: True if left child exists.

has_right_child ¶

has_right_child() -> bool

Check if node has right child.

bool: True if right child exists.

inorder_traversal ¶

inorder_traversal() -> list[BinaryTreeNode]

Perform in-order traversal (left, root, right).

list[Any]: Values in in-order.

is_leaf ¶

is_leaf() -> bool

Check if node is a leaf.

bool: True if node has no children.

postorder_traversal ¶

postorder_traversal() -> list[BinaryTreeNode]

Perform post-order traversal (left, right, root).

list[Any]: Values in post-order.

preorder_traversal ¶

preorder_traversal() -> list[BinaryTreeNode]

Perform pre-order traversal (root, left, right).

list[Any]: Values in pre-order.

traverse_inorder ¶

traverse_inorder() -> list[BinaryTreeNode]

Alias for inorder_traversal for compatibility.

traverse_postorder ¶

traverse_postorder() -> list[BinaryTreeNode]

Alias for postorder_traversal for compatibility.

traverse_preorder ¶

traverse_preorder() -> list[BinaryTreeNode]

Alias for preorder_traversal for compatibility.

nested_set ¶

Nested Set Structure Module¶

This module provides a NestedSetStructure class that allows for the management of hierarchical relationships between items. It supports adding items with parent-child relationships, retrieving children and parents, and provides a way to retrieve all items in a nested, flattened order.

Classes¶

NestedSetStructure ¶

NestedSetStructure()

Nested Set Structure Class¶

A structure to track items and their hierarchical relationships. Supports multi-level trees with parent/child tracking. Useful for scenarios like recursive publishing, dependency resolution, or graph-like traversal.

Initialize the nested set structure.

Functions¶

add ¶

add(item: Any, parent: Any | None = None) -> None

Add an item to the structure, optionally specifying a parent.

children ¶

children(item: Any) -> list[Any]

Return the children of the given item. If the item has no children, returns an empty list.

nested_items ¶

nested_items() -> list[Any]

Return a nested, flattened list of items (including children) in insertion order. Maintains hierarchy order.

original ¶

original(item: Any) -> Any

Return the original object stored in the set that equals item, or item itself if not found. This is useful for matching back to canonical instances.

parent ¶

parent(item: Any) -> Any | None

Return the parent of the given item, or None if it has no parent.

tree_node ¶

Tree Node¶

Generic tree node implementation with children and parent tracking.

Classes¶

TreeNode ¶

TreeNode(value: Any, children: list[Self] | None = None)

TreeNode Class¶

A generic tree node that can have multiple children.

Parameters¶

value : Any The value stored in this node. children : list[TreeNode] | None Optional list of child nodes.

Initialize a tree node.

value: The value to store in this node.
children: Optional list of child nodes.

Functions¶

add_child ¶

add_child(child: Self) -> None

Add a child node.

child: The child node to add.

get_ancestors ¶

get_ancestors() -> list[Self]

Get all ancestor nodes from parent to root.

list[TreeNode]: List of ancestors in order (parent to root).

get_depth ¶

get_depth() -> int

Get the depth of this node (distance from root).

int: Depth of node (root is 0).

get_height ¶

get_height() -> int

Get the height of the subtree rooted at this node.

int: Height of subtree (leaf is 0).

get_siblings ¶

get_siblings() -> list[Self]

Get all sibling nodes (nodes with same parent).

list[TreeNode]: List of sibling nodes.

is_leaf ¶

is_leaf() -> bool

Check if this node is a leaf (has no children).

bool: True if node has no children.

is_root ¶

is_root() -> bool

Check if this node is a root (has no parent).

bool: True if node has no parent.

remove_child ¶

remove_child(child: Self) -> bool

Remove a child node.

child: The child node to remove.

bool: True if child was removed, False if not found.

traverse_levelorder ¶

traverse_levelorder() -> list[Self]

Traverse tree in level-order (breadth-first).

list[TreeNode]: Nodes in level-order.

traverse_postorder ¶

traverse_postorder() -> list[Self]

Traverse tree in post-order (children, then root).

list[TreeNode]: Nodes in post-order.

traverse_preorder ¶

traverse_preorder() -> list[Self]

Traverse tree in pre-order (root, then children).

list[TreeNode]: Nodes in pre-order.

trie ¶

Trie (Prefix Tree)¶

A trie data structure for efficient string prefix operations.

Classes¶

Trie ¶

Trie()

Trie Class¶

A prefix tree (trie) for efficient string operations like autocomplete, spell checking, and prefix matching.

Initialize an empty trie.

Functions¶

autocomplete ¶

autocomplete(prefix: str) -> list[str]

Return all completions for the given prefix.

delete ¶

delete(word: str) -> bool

Delete a word from the trie.

word: The word to delete.

bool: True if word was deleted, False if not found.

get ¶

get(word: str) -> Any | None

Get the value associated with a word.

word: The word to look up.

Any | None: Associated value or None if not found.

get_all_words ¶

get_all_words() -> list[str]

Get all words stored in the trie.

list[str]: All words in the trie.

get_words_with_prefix ¶

get_words_with_prefix(prefix: str) -> list[str]

Get all words that start with given prefix.

prefix: The prefix to match.

list[str]: List of words with the prefix.

insert ¶

insert(word: str, value: Any | None = None) -> None

Insert a word into the trie.

word: The word to insert.
value: Optional value to associate with the word.

search ¶

search(word: str) -> bool

Search for an exact word in the trie.

word: The word to search for.

bool: True if word exists in trie.

starts_with ¶

starts_with(prefix: str) -> bool

Check if any word in trie starts with given prefix.

prefix: The prefix to check.

bool: True if prefix exists in trie.

TrieNode ¶

TrieNode()

TrieNode Class¶

A single node in a Trie structure.

Initialize a trie node.

Submodules¶

Cache¶

LRU, LFU, and TTL cache implementations.

Cache Collections Module¶

Provides caching data structures with various eviction policies.

Classes:¶

LRUCache: Least Recently Used cache
LFUCache: Least Frequently Used cache
TTLCache: Time-To-Live cache

Modules¶

lru_cache ¶

LRU Cache¶

Least Recently Used (LRU) cache implementation.

Classes¶

LRUCache ¶

LRUCache(capacity: int)

LRUCache Class¶

A Least Recently Used (LRU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LRU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

items ¶

items() -> list[tuple[Any, Any]]

Get all key-value pairs.

list[tuple[Any, Any]]: List of (key, value) tuples.

keys ¶

keys() -> list[Any]

Get all keys in cache.

list[Any]: List of keys in order of use.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

values ¶

values() -> list[Any]

Get all values in cache.

list[Any]: List of values.

lfu_cache ¶

LFU Cache¶

Least Frequently Used (LFU) cache implementation.

Classes¶

LFUCache ¶

LFUCache(capacity: int)

LFUCache Class¶

A Least Frequently Used (LFU) cache with fixed capacity.

Parameters¶

capacity : int Maximum number of items to store in cache.

Initialize an LFU cache.

capacity: Maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache.

key: The key to look up.

Any | None: The value or None if not found.

put ¶

put(key: Any, value: Any) -> None

Add or update a key-value pair in cache.

key: The key to store.
value: The value to store.

ttl_cache ¶

TTL Cache¶

Time-To-Live (TTL) cache with automatic expiration.

Classes¶

TTLCache ¶

TTLCache(default_ttl: float = 300.0, max_size: int | None = None)

TTLCache Class¶

A cache with Time-To-Live (TTL) expiration for entries.

Parameters¶

default_ttl : float Default time-to-live in seconds. max_size : int | None Optional maximum cache size.

Initialize a TTL cache.

default_ttl: Default expiration time in seconds.
max_size: Optional maximum cache size.

Functions¶

clear ¶

clear() -> None

Clear all items from cache.

delete ¶

delete(key: Any) -> bool

Delete a key from cache.

key: The key to delete.

bool: True if key was deleted, False if not found.

get ¶

get(key: Any) -> Any | None

Get a value from cache if not expired.

key: The key to look up.

Any | None: The value or None if not found/expired.

get_ttl ¶

get_ttl(key: Any) -> float | None

Get remaining TTL for a key.

key: The key to check.

float | None: Remaining seconds or None if not found/expired.

put ¶

put(key: Any, value: Any, ttl: float | None = None) -> None

Add or update a key-value pair with TTL.

key: The key to store.
value: The value to store.
ttl: Optional TTL in seconds (uses default if not provided).

Buffer¶

Circular and sliding window buffers.

Buffer Collections Module¶

Provides various buffer data structures for specialized data storage patterns.

Classes:¶

CircularBuffer: Ring buffer with automatic overwriting
RingBuffer: Alias for CircularBuffer
BoundedBuffer: Size-limited buffer with overflow protection
SlidingWindow: Moving window over a data stream

Modules¶

circular_buffer ¶

Circular Buffer¶

A circular buffer (ring buffer) implementation that overwrites the oldest entries when the buffer reaches its maximum capacity.

Classes:¶

CircularBuffer: A fixed-size buffer with circular indexing.

Features:¶

Append new elements, overwriting the oldest if the buffer is full.
Retrieve all elements in the correct order.
Check if the buffer is empty or full.

Usage:¶

buffer = CircularBuffer(size=5)
buffer.append(1)
buffer.append(2)
print(buffer.get_all())  # Outputs: [1, 2]
buffer.append(3)
buffer.append(4)
buffer.append(5)
print(buffer.get_all())  # Outputs: [1, 2, 3, 4, 5]
buffer.append(6)
print(buffer.get_all())  # Outputs: [6, 2, 3, 4, 5]

Classes¶

CircularBuffer ¶

CircularBuffer(size: int)

CircularBuffer Class¶

A fixed-size buffer that overwrites the oldest data when full.

Attributes¶

size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.

Methods¶

append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.

Initializes a CircularBuffer instance.

Parameters:¶

size : int The maximum number of elements the buffer can hold.

Functions¶

append ¶

append(value: Any) -> None

Adds a value to the buffer. Overwrites the oldest element if the buffer is full.

Parameters:¶

value : Any The value to add to the buffer.

clear ¶

clear() -> None

Clear the buffer.

get ¶

get(index: int) -> Any | None

Return item at index or None if out of bounds.

get_all ¶

get_all() -> list[Any | None]

Retrieves all elements in the buffer in the correct order.

Returns¶

list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.

is_empty ¶

is_empty() -> bool

Checks if the buffer is empty.

Returns¶

bool: True if the buffer is empty, False otherwise.

is_full ¶

is_full() -> bool

Checks if the buffer is full.

Returns¶

bool: True if the buffer is full, False otherwise.

bounded_buffer ¶

Bounded Buffer¶

A thread-safe bounded buffer that blocks or raises errors when capacity limits are reached.

Classes¶

BoundedBuffer ¶

BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')

BoundedBuffer Class¶

A size-limited buffer that prevents overflow with configurable behavior.

Parameters¶

maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'

Initialize the bounded buffer.

maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.

Attributes¶

capacity property ¶

capacity: int

Return maximum number of elements the buffer can hold.

Functions¶

append ¶

append(item: Any) -> bool

Add an item to the buffer.

item: The item to add.

bool: True if item was added, False if dropped.

BufferError: If overflow_strategy is 'raise' and buffer is full.

clear ¶

clear() -> None

Clear all items from buffer.

extend ¶

extend(items: list[Any]) -> None

Extend buffer with multiple items.

items: List of items to add.

get ¶

get(index: int) -> Any | None

Return the item at a given index or None if out of bounds.

get_all ¶

get_all() -> list[Any]

Get all items in buffer.

list[Any]: All items in insertion order.

is_empty ¶

is_empty() -> bool

Check if buffer is empty.

bool: True if buffer contains no items.

is_full ¶

is_full() -> bool

Check if buffer is full.

bool: True if buffer has reached maxsize.

peek ¶

peek() -> Any | None

Return the oldest item without removing it.

ring_buffer ¶

Ring Buffer¶

Alias for CircularBuffer providing ring buffer functionality.

Classes¶

sliding_window ¶

Sliding Window¶

A sliding window data structure for stream processing and moving calculations.

Classes¶

SlidingWindow ¶

SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)

SlidingWindow Class¶

A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.

Parameters¶

size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.

Initialize the sliding window.

size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.

Functions¶

add ¶

add(value: Any) -> Any | None

Add a value to the window and return aggregated result.

value: The value to add to the window.

Any | None: Aggregated result if aggregation_func is set.

clear ¶

clear() -> None

Clear all elements from window.

get_aggregate ¶

get_aggregate() -> Any | None

Get aggregated value of current window.

Any | None: Aggregated result or None if no aggregation function.

get_window ¶

get_window() -> list[Any]

Get current window contents.

list[Any]: Current window values in order.

is_empty ¶

is_empty() -> bool

Check if window is empty.

bool: True if window contains no elements.

is_full ¶

is_full() -> bool

Check if window is full.

bool: True if window has reached its size.

moving_average ¶

moving_average() -> float | None

Calculate moving average of numeric window.

float | None: Average of window values or None if empty.

moving_max ¶

moving_max() -> Any | None

Get maximum value in window.

Any | None: Maximum value or None if empty.

moving_min ¶

moving_min() -> Any | None

Get minimum value in window.

Any | None: Minimum value or None if empty.

moving_sum ¶

moving_sum() -> float

Calculate moving sum of numeric window.

float | None: Sum of window values or None if empty.

Dictionary Utilities¶

Deep operations on dictionaries.

Dictionary Operations Module¶

Utilities for dictionary manipulation and transformation.

Functions¶

dict_merge: Merge multiple dictionaries.
dict_filter: Filter dictionary by predicate.
dict_invert: Swap keys and values.
dict_deep_get: Get nested value safely.
dict_deep_set: Set nested value, creating paths.

Examples¶

from rite.collections.dict import dict_merge, dict_filter dict_merge({"a": 1}, {"b": 2}) {'a': 1, 'b': 2} dict_filter({"a": 1, "b": 2}, lambda k, v: v > 1)

Modules¶

dict_deep_get ¶

Dict Deep Get¶

Safely get nested dictionary values using key paths.

Functions¶

dict_deep_get: Get nested value using list of keys.

Examples¶

from rite.collections.dict import dict_deep_get data = {"user": {"profile": {"name": "John"}}} dict_deep_get(data, ["user", "profile", "name"]) 'John'

Functions¶

dict_deep_get ¶

dict_deep_get(d: dict[str, Any], keys: list[str], default: Any = None) -> Any

Get nested dictionary value using key path.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to traverse.	required
`keys`	`list[str]`	List of keys forming path to desired value.	required
`default`	`Any`	Value to return if path doesn't exist.	`None`

Returns:

Type	Description
`Any`	Value at the key path or default.

Examples:

>>> data = {"user": {"profile": {"name": "John"}}}
>>> dict_deep_get(data, ["user", "profile", "name"])
'John'
>>> dict_deep_get(data, ["user", "missing"], "N/A")
'N/A'
>>> dict_deep_get({}, ["a", "b"])
None

dict_deep_set ¶

Dict Deep Set¶

Set nested dictionary values, creating intermediate dicts as needed.

Functions¶

dict_deep_set: Set value at nested key path.

Examples¶

from rite.collections.dict import dict_deep_set d = {} dict_deep_set(d, ["user", "profile", "name"], "John") d {'user': {'profile': {'name': 'John'}}}

Functions¶

dict_deep_set ¶

dict_deep_set(d: dict[str, Any], keys: list[str], value: Any) -> None

Set nested dictionary value, creating intermediate dicts.

Modifies dictionary in-place.

Parameters:

Name	Type	Description	Default
`d`	`dict[str, Any]`	Dictionary to modify.	required
`keys`	`list[str]`	List of keys forming path where value should be set.	required
`value`	`Any`	Value to set at the key path.	required

Examples:

>>> d = {}
>>> dict_deep_set(d, ["user", "profile", "name"], "John")
>>> d
{'user': {'profile': {'name': 'John'}}}
>>> dict_deep_set(d, ["user", "age"], 30)
>>> d['user']['age']
30

dict_merge ¶

Dict Merge¶

Merge multiple dictionaries with various strategies.

Functions¶

dict_merge: Merge dictionaries with configurable behavior.

Examples¶

from rite.collections.dict import dict_merge dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})

Functions¶

dict_merge ¶

dict_merge(*dicts: dict[K, V], deep: bool = False) -> dict[K, V]

Merge multiple dictionaries.

Later dictionaries override earlier ones for duplicate keys.

Parameters:

Name	Type	Description	Default
`*dicts`	`dict[K, V]`	Variable number of dictionaries to merge.	`()`
`deep`	`bool`	If True, recursively merge nested dicts.	`False`

Returns:

Type	Description
`dict[K, V]`	New merged dictionary.

Examples:

>>> dict_merge({"a": 1, "b": 2}, {"b": 3, "c": 4})
{'a': 1, 'b': 3, 'c': 4}
>>> dict_merge({"a": 1}, {"b": 2}, {"c": 3})
{'a': 1, 'b': 2, 'c': 3}
>>> d1 = {"a": {"x": 1}}
>>> d2 = {"a": {"y": 2}}
>>> dict_merge(d1, d2, deep=True)
{'a': {'x': 1, 'y': 2}}

dict_filter ¶

Dict Filter¶

Filter dictionary entries based on key or value predicates.

Functions¶

dict_filter: Filter dictionary by predicate.

Examples¶

from rite.collections.dict import dict_filter dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)

Functions¶

dict_filter ¶

dict_filter(d: dict[K, V], predicate: Callable[[K, V], bool]) -> dict[K, V]

Filter dictionary entries based on predicate function.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to filter.	required
`predicate`	`Callable[[K, V], bool]`	Function taking (key, value) returning True to keep.	required

Returns:

Type	Description
`dict[K, V]`	New dictionary with filtered entries.

Examples:

>>> dict_filter({"a": 1, "b": 2, "c": 3}, lambda k, v: v > 1)
{'b': 2, 'c': 3}
>>> dict_filter({"a": 1, "b": 2}, lambda k, v: k == "a")
{'a': 1}
>>> dict_filter({}, lambda k, v: True)
{}

dict_invert ¶

Dict Invert¶

Invert dictionary keys and values.

Functions¶

dict_invert: Swap dictionary keys and values.

Examples¶

from rite.collections.dict import dict_invert dict_invert({"a": 1, "b": 2})

Functions¶

dict_invert ¶

dict_invert(d: dict[K, V]) -> dict[V, K]

Invert dictionary keys and values.

Parameters:

Name	Type	Description	Default
`d`	`dict[K, V]`	Dictionary to invert.	required

Returns:

Type	Description
`dict[V, K]`	New dictionary with keys and values swapped.

Raises:

Type	Description
`TypeError`	If values are not hashable.

Examples:

>>> dict_invert({"a": 1, "b": 2})
{1: 'a', 2: 'b'}
>>> dict_invert({1: "one", 2: "two"})
{'one': 1, 'two': 2}
>>> dict_invert({})
{}

List Utilities¶

Advanced list operations.

List Operations Module¶

Utilities for list manipulation and transformation.

Functions¶

list_unique: Remove duplicates while preserving order.
list_flatten: Flatten nested lists recursively.
list_chunk: Split list into chunks.
list_group_by: Group items by key function or attribute.
list_partition: Split list into two based on predicate.
list_interleave: Interleave multiple lists.

Examples¶

from rite.collections.list import list_unique, list_chunk list_unique([1, 2, 2, 3]) [1, 2, 3] list_chunk([1, 2, 3, 4, 5], size=2) [[1, 2], [3, 4], [5]]

Modules¶

list_chunk ¶

List Chunk¶

Split a list into chunks of specified size.

Functions¶

list_chunk: Split list into smaller chunks.

Examples¶

from rite.collections.list import list_chunk list_chunk([1, 2, 3, 4, 5], size=2) [[1, 2], [3, 4], [5]] list_chunk([1, 2, 3, 4, 5, 6], size=3) [[1, 2, 3], [4, 5, 6]]

Functions¶

list_chunk ¶

list_chunk(items: list[T], size: int) -> list[list[T]]

Split a list into chunks of specified size.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to split into chunks.	required
`size`	`int`	Size of each chunk. Must be positive.	required

Returns:

Type	Description
`list[list[T]]`	List of chunks, last chunk may be smaller.

Raises:

Type	Description
`ValueError`	If size is not positive.

Examples:

>>> list_chunk([1, 2, 3, 4, 5], size=2)
[[1, 2], [3, 4], [5]]
>>> list_chunk([1, 2, 3, 4, 5, 6], size=3)
[[1, 2, 3], [4, 5, 6]]
>>> list_chunk([], size=2)
[]
>>> list_chunk([1, 2, 3], size=5)
[[1, 2, 3]]

list_flatten ¶

List Flatten¶

Flatten nested lists into a single-level list.

Functions¶

list_flatten: Flatten nested lists recursively.

Examples¶

from rite.collections.list import list_flatten list_flatten([[1, 2], [3, 4], [5]]) [1, 2, 3, 4, 5] list_flatten([[1, [2, 3]], [4, 5]]) [1, 2, 3, 4, 5]

Functions¶

list_flatten ¶

list_flatten(items: list[Any], depth: int | None = None) -> list[Any]

Flatten nested lists recursively.

Parameters:

Name	Type	Description	Default
`items`	`list[Any]`	List that may contain nested lists.	required
`depth`	`int \| None`	Maximum depth to flatten. None for full recursion.	`None`

Returns:

Type	Description
`list[Any]`	Flattened list.

Examples:

>>> list_flatten([[1, 2], [3, 4], [5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, 3]], [4, 5]])
[1, 2, 3, 4, 5]
>>> list_flatten([[1, [2, [3]]], [4]], depth=1)
[1, [2, [3]], 4]
>>> list_flatten([])
[]

list_group_by ¶

List Group By¶

Group list items by a key function or attribute.

Functions¶

list_group_by: Group items by key.

Examples¶

from rite.collections.list import list_group_by data = [{"type": "fruit", "name": "apple"}] list_group_by(data, key="type") {'fruit': [{'type': 'fruit', 'name': 'apple'}]}

Functions¶

list_group_by ¶

list_group_by(items: list[T], key: str | Callable[[T], K]) -> dict[Any, list[T]]

Group list items by a key function or dict key.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List of items to group.	required
`key`	`str \| Callable[[T], K]`	Either attribute name (str) or function to extract key.	required

Returns:

Type	Description
`dict[Any, list[T]]`	Dictionary mapping keys to lists of items.

Examples:

>>> data = [
...     {"type": "fruit", "name": "apple"},
...     {"type": "fruit", "name": "banana"},
...     {"type": "veg", "name": "carrot"}
... ]
>>> list_group_by(data, key="type")
{'fruit': [...], 'veg': [...]}
>>> list_group_by([1, 2, 3, 4], key=lambda x: x % 2)
{1: [1, 3], 0: [2, 4]}

list_interleave ¶

List Interleave¶

Interleave multiple lists together.

Functions¶

list_interleave: Combine lists by alternating elements.

Examples¶

from rite.collections.list import list_interleave list_interleave([1, 2, 3], ['a', 'b', 'c']) [1, 'a', 2, 'b', 3, 'c']

Functions¶

list_interleave ¶

list_interleave(*lists: list[Any]) -> list[Any]

Interleave multiple lists by alternating elements.

Parameters:

Name	Type	Description	Default
`*lists`	`list[Any]`	Variable number of lists to interleave.	`()`

Returns:

Type	Description
`list[Any]`	New list with elements from all lists interleaved.

Examples:

>>> list_interleave([1, 2, 3], ['a', 'b', 'c'])
[1, 'a', 2, 'b', 3, 'c']
>>> list_interleave([1, 2], [10, 20], [100, 200])
[1, 10, 100, 2, 20, 200]
>>> list_interleave([1, 2, 3], ['a'])
[1, 'a', 2, 3]
>>> list_interleave([])
[]

list_partition ¶

List Partition¶

Partition a list into two lists based on a predicate function.

Functions¶

list_partition: Split list into matching and non-matching items.

Examples¶

from rite.collections.list import list_partition list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0) ([2, 4], [1, 3, 5])

Functions¶

list_partition ¶

list_partition(items: list[T], predicate: Callable[[T], bool]) -> tuple[list[T], list[T]]

Partition a list based on a predicate function.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List to partition.	required
`predicate`	`Callable[[T], bool]`	Function that returns True for items in first list.	required

Returns:

Type	Description
`tuple[list[T], list[T]]`	Tuple of (matching_items, non_matching_items).

Examples:

>>> list_partition([1, 2, 3, 4, 5], lambda x: x % 2 == 0)
([2, 4], [1, 3, 5])
>>> list_partition(['a', 'bb', 'ccc'], lambda x: len(x) > 1)
(['bb', 'ccc'], ['a'])
>>> list_partition([], lambda x: True)
([], [])

list_unique ¶

List Unique¶

Remove duplicate elements from a list while preserving order.

Functions¶

list_unique: Remove duplicates from a list.

Examples¶

from rite.collections.list import list_unique list_unique([1, 2, 2, 3, 3, 3]) [1, 2, 3] list_unique(['a', 'b', 'a', 'c']) ['a', 'b', 'c']

Functions¶

list_unique ¶

list_unique(items: list[T]) -> list[T]

Remove duplicate elements while preserving order.

Uses a set to track seen elements and preserves first occurrence of each unique element.

Parameters:

Name	Type	Description	Default
`items`	`list[T]`	List that may contain duplicates.	required

Returns:

Type	Description
`list[T]`	New list with duplicates removed, order preserved.

Examples:

>>> list_unique([1, 2, 2, 3, 3, 3])
[1, 2, 3]
>>> list_unique(['a', 'b', 'a', 'c'])
['a', 'b', 'c']
>>> list_unique([])
[]

Tree Structures¶

Tree and trie data structures.

Tree Collections Module¶

Provides tree data structures and hierarchical data management.

Classes:¶

NestedSetStructure: Hierarchical parent-child relationship manager
TreeNode: Generic tree node with children
BinaryTreeNode: Binary tree node implementation
Trie: Prefix tree for string operations

Classes¶

TreeNode ¶

TreeNode(value: Any, children: list[Self] | None = None)

TreeNode Class¶

A generic tree node that can have multiple children.

Parameters¶

value : Any The value stored in this node. children : list[TreeNode] | None Optional list of child nodes.

Initialize a tree node.

value: The value to store in this node.
children: Optional list of child nodes.

Functions¶

add_child ¶

add_child(child: Self) -> None

Add a child node.

child: The child node to add.

get_ancestors ¶

get_ancestors() -> list[Self]

Get all ancestor nodes from parent to root.

list[TreeNode]: List of ancestors in order (parent to root).

get_depth ¶

get_depth() -> int

Get the depth of this node (distance from root).

int: Depth of node (root is 0).

get_height ¶

get_height() -> int

Get the height of the subtree rooted at this node.

int: Height of subtree (leaf is 0).

get_siblings ¶

get_siblings() -> list[Self]

Get all sibling nodes (nodes with same parent).

list[TreeNode]: List of sibling nodes.

is_leaf ¶

is_leaf() -> bool

Check if this node is a leaf (has no children).

bool: True if node has no children.

is_root ¶

is_root() -> bool

Check if this node is a root (has no parent).

bool: True if node has no parent.

remove_child ¶

remove_child(child: Self) -> bool

Remove a child node.

child: The child node to remove.

bool: True if child was removed, False if not found.

traverse_levelorder ¶

traverse_levelorder() -> list[Self]

Traverse tree in level-order (breadth-first).

list[TreeNode]: Nodes in level-order.

traverse_postorder ¶

traverse_postorder() -> list[Self]

Traverse tree in post-order (children, then root).

list[TreeNode]: Nodes in post-order.

traverse_preorder ¶

traverse_preorder() -> list[Self]

Traverse tree in pre-order (root, then children).

list[TreeNode]: Nodes in pre-order.

BinaryTreeNode ¶

BinaryTreeNode(value: Any, left: BinaryTreeNode | None = None, right: BinaryTreeNode | None = None)

BinaryTreeNode Class¶

A binary tree node with left and right children.

Parameters¶

value : Any The value stored in this node. left : BinaryTreeNode | None Left child node. right : BinaryTreeNode | None Right child node.

Initialize a binary tree node.

value: The value to store.
left: Left child node.
right: Right child node.

Functions¶

get_height ¶

get_height() -> int

Get height of subtree rooted at this node.

int: Height of subtree.

has_left_child ¶

has_left_child() -> bool

Check if node has left child.

bool: True if left child exists.

has_right_child ¶

has_right_child() -> bool

Check if node has right child.

bool: True if right child exists.

inorder_traversal ¶

inorder_traversal() -> list[BinaryTreeNode]

Perform in-order traversal (left, root, right).

list[Any]: Values in in-order.

is_leaf ¶

is_leaf() -> bool

Check if node is a leaf.

bool: True if node has no children.

postorder_traversal ¶

postorder_traversal() -> list[BinaryTreeNode]

Perform post-order traversal (left, right, root).

list[Any]: Values in post-order.

preorder_traversal ¶

preorder_traversal() -> list[BinaryTreeNode]

Perform pre-order traversal (root, left, right).

list[Any]: Values in pre-order.

traverse_inorder ¶

traverse_inorder() -> list[BinaryTreeNode]

Alias for inorder_traversal for compatibility.

traverse_postorder ¶

traverse_postorder() -> list[BinaryTreeNode]

Alias for postorder_traversal for compatibility.

traverse_preorder ¶

traverse_preorder() -> list[BinaryTreeNode]

Alias for preorder_traversal for compatibility.

Trie ¶

Trie()

Trie Class¶

A prefix tree (trie) for efficient string operations like autocomplete, spell checking, and prefix matching.

Initialize an empty trie.

Functions¶

autocomplete ¶

autocomplete(prefix: str) -> list[str]

Return all completions for the given prefix.

delete ¶

delete(word: str) -> bool

Delete a word from the trie.

word: The word to delete.

bool: True if word was deleted, False if not found.

get ¶

get(word: str) -> Any | None

Get the value associated with a word.

word: The word to look up.

Any | None: Associated value or None if not found.

get_all_words ¶

get_all_words() -> list[str]

Get all words stored in the trie.

list[str]: All words in the trie.

get_words_with_prefix ¶

get_words_with_prefix(prefix: str) -> list[str]

Get all words that start with given prefix.

prefix: The prefix to match.

list[str]: List of words with the prefix.

insert ¶

insert(word: str, value: Any | None = None) -> None

Insert a word into the trie.

word: The word to insert.
value: Optional value to associate with the word.

search ¶

search(word: str) -> bool

Search for an exact word in the trie.

word: The word to search for.

bool: True if word exists in trie.

starts_with ¶

starts_with(prefix: str) -> bool

Check if any word in trie starts with given prefix.

prefix: The prefix to check.

bool: True if prefix exists in trie.

Examples¶

from rite.collections import (
    lru_cache,
    circular_buffer,
    dict_deep_get,
    list_chunk
)

# LRU Cache
cache = lru_cache(maxsize=100)

# Circular Buffer
buffer = circular_buffer(capacity=10)
buffer.append(1)
buffer.append(2)

# Deep dictionary access
data = {"a": {"b": {"c": 42}}}
value = dict_deep_get(data, ["a", "b", "c"])  # 42

# Chunk list
items = [1, 2, 3, 4, 5, 6]
chunks = list_chunk(items, 2)  # [[1, 2], [3, 4], [5, 6]]

Collections Module¶

Overview¶

collections ¶