Documentation
Function Utilities
Function composition, throttling, debouncing, and more.
Tools for function manipulation, currying, throttling, and more.
partial()
Creates a partially applied function
Implementation
Returns a new function with some arguments pre-filled
Example
add_five = UniCoreFW.partial(lambda a, b: a + b, 5)
Expected output: Function that adds 5 to its argument
Source Code
def partial(func, *partial_args):
def partially_applied(*extra_args):
return func(*(partial_args + extra_args))
return partially_applied
unary()
Create a function that accepts only the first argument, ignoring others.
Implementation
Args: func: The function to cap arguments for
Returns:
A function that accepts only one argument
Example
def add_all(*args):
return sum(args)
first_only = unary(add_all)
first_only(1, 2, 3, 4) # Only uses first argument
Expected output: 1
Source Code
def unary(func: Callable) -> Callable:
_validate_callable(func)
def wrapper(arg, *ignored_args, **kwargs):
return func(arg, **kwargs)
return _copy_function_metadata(wrapper, func)
spread()
Create a function that spreads an array argument to func as individual arguments.
Implementation
Args: func: The function to spread arguments to
Returns:
A function that spreads array arguments
Example
def add_three(a, b, c):
return a + b + c
spread_add = spread(add_three)
spread_add([1, 2, 3])
Expected output: 6
Source Code
def spread(func: Callable) -> Callable:
_validate_callable(func)
def wrapper(args_array, **kwargs):
return func(*args_array, **kwargs)
return _copy_function_metadata(wrapper, func)
over_args()
Create a function that applies transformations to arguments before calling func.
Implementation
Args: func: The function to invoke *transforms: Functions to transform corresponding arguments
Returns:
A function that transforms arguments before calling func
Example
def add(a, b):
return a + b
square = lambda x: x * x
double = lambda x: x * 2
transform_add = over_args(add, square, double)
transform_add(3, 4) # (3*3) + (4*2) = 9 + 8 = 17
Expected output: 17
Source Code
def over_args(func: Callable, *transforms: Callable) -> Callable:
_validate_callable(func)
# Handle case where transforms are passed as a single tuple/list
if len(transforms) == 1 and isinstance(transforms[0], (list, tuple)):
transforms = transforms[0] # type: ignore
for transform in transforms:
_validate_callable(transform)
def wrapper(*args, **kwargs):
transformed_args = []
for i, arg in enumerate(args):
if i < len(transforms):
transformed_args.append(transforms[i](arg))
else:
transformed_args.append(arg)
return func(*transformed_args, **kwargs)
return _copy_function_metadata(wrapper, func)
debounce_()
Enhanced debounce with max_wait support.
Implementation
Implementation details are not available.
Example
Expected output: ``
Source Code
def debounce_(func: Callable, wait: int, *, max_wait: Optional[int] = None, use_main_thread: bool = False) -> Callable:
if max_wait is not None:
return enhanced_debounce(func, wait, max_wait)
else:
# Use original implementation for backward compatibility
return enhanced_debounce(func, wait)
partial_right()
Enhanced partial_right with _argcount support.
Implementation
Implementation details are not available.
Example
Expected output: ``
Source Code
def partial_right(func: Callable, *bound_args, **bound_kwargs) -> Callable:
return enhanced_partial_right(func, *bound_args, **bound_kwargs)
throttle()
Limits a function's execution rate
Implementation
Tracks the last call time and only calls if enough time has passed
Example
throttled = UniCoreFW.throttle(lambda: print('Called'), 1000)
Expected output: Function that executes at most once per second
Source Code
def throttle(func, wait):
import time
last_called = [0]
def throttled_func(*args, **kwargs):
now = time.time()
if now - last_called[0] > wait / 1000:
last_called[0] = now
return func(*args, **kwargs)
return throttled_func
conjoin()
Returns a function that tests whether all elements of a sequence pass all of the provided predicates.
Implementation
Args: *preds: Functions to conjoin.
Returns:
A function that takes an iterable, and returns True if all elements of
the iterable pass all predicates, False otherwise.
Example
is_even = lambda x: x % 2 == 0
is_positive = lambda x: x > 0
both = conjoin(is_even, is_positive)
both([2, 4, 6])
Expected output: True
Source Code
def conjoin(*preds):
def wrapper(seq):
return all(all(p(x) for p in preds) for x in seq)
return wrapper
rearg()
Creates a function that invokes func with arguments rearranged according to the specified
indexes. The argument value at the first index is provided as the first argument, the argument
value at the second index is provided as the second argument, and so on. Arguments not specified
in indexes maintain their original order after the rearranged arguments.
Implementation
Args: func: Function to rearrange arguments for. *indexes: The specified argument indexes.
Returns:
A new function with rearranged arguments for `func`.
Example
def example(a, b, c):
return a, b, c
rearranged = rearg(example, 2, 0, 1)
rearranged(1, 2, 3)
Expected output: (3, 1, 2)
Source Code
def rearg(func: Callable[..., R], *indexes: int) -> Callable[..., R]:
# Normalize a single list/tuple argument
if len(indexes) == 1 and isinstance(indexes[0], (list, tuple)):
idxs = tuple(indexes[0])
else:
idxs = indexes
def wrapper(*args: Any, **kwargs: Any) -> R:
# Pick valid indices in the order given
valid = [i for i in idxs if isinstance(i, int) and 0 <= i < len(args)]
# Reordered portion
reordered = [args[i] for i in valid]
# Append the args not yet used
remaining = [args[i] for i in range(len(args)) if i not in valid]
return func(*reordered, *remaining, **kwargs)
# Copy metadata for introspection
wrapper.__name__ = func.__name__
wrapper.__doc__ = func.__doc__
wrapper.__wrapped__ = func # type: ignore
return wrapper
debounce()
Delays execution until pause in calls
Implementation
Uses a timer that resets on each call
Example
debounced = UniCoreFW.debounce(lambda: print('Called'), 200)
Expected output: Function that executes 200ms after the last call
Source Code
def debounce(func, wait, *, use_main_thread=False):
wait_seconds = wait / 1000.0
timer = None
lock = threading.Lock()
def debounced(*args, **kwargs):
nonlocal timer
def call_func():
if use_main_thread:
# Queue the function to be called in the main thread
threading.main_thread().run(func, *args, **kwargs)
else:
func(*args, **kwargs)
with lock:
if timer is not None:
timer.cancel()
timer = threading.Lock()
timer = threading.Timer(wait_seconds, call_func)
timer.start()
return debounced
once()
Creates a function that executes only once
Implementation
Uses a closure to track if the function has been called
Example
init_once = UniCoreFW.once(lambda: print('Initialized'))
Expected output: Function that only prints on the first call
Source Code
def once(func):
result = [None]
has_been_called = [False]
def once_func(*args, **kwargs):
if not has_been_called[0]:
result[0] = func(*args, **kwargs)
has_been_called[0] = True
else:
result[0] = None # Return None after the first call
return result[0]
return once_func
after()
Creates a function that executes after n calls
Implementation
Tracks the number of calls in a closure
Example
after_three = UniCoreFW.after(3, lambda: print('Now ready'))
Expected output: Function that prints only after being called 3 times
Source Code
def after(times, func):
calls = [0]
def after_func(*args, **kwargs):
calls[0] += 1
if calls[0] >= times:
return func(*args, **kwargs)
return after_func
filter_()
Filter implementation for test compatibility.
Implementation
Args: iterable: The collection to filter predicate: The filter function
Returns:
List of filtered values
Example
filter_([1, 2, 3], lambda x: x % 2 == 0)
Expected output: [2]
Source Code
def filter_(iterable, predicate):
return [item for item in iterable if predicate(item)]
map_()
Map implementation for test compatibility.
Implementation
Args: iterable: The collection to map over func: The mapping function
Returns:
List of mapped values
Example
map_([1, 2, 3], lambda x: x * x)
Expected output: [1, 4, 9]
Source Code
def map_(iterable, func):
return [func(item) for item in iterable]
reduce_()
Reduce implementation for test compatibility.
Implementation
Args: iterable: The collection to reduce func: The reducer function initial: Initial value
Returns:
The reduced value
Example
reduce_([1, 2, 3], lambda x, y: x + y)
Expected output: 6
Source Code
def reduce_(iterable, func, initial=None):
if initial is None:
iterator = iter(iterable)
try:
initial = next(iterator)
except StopIteration:
raise TypeError("reduce() of empty sequence with no initial value")
iterable = iterator
result = initial
for item in iterable:
result = func(result, item)
return result
before()
Creates a function that executes at most n times
Implementation
Similar to once but with a configurable count
Example
only_twice = UniCoreFW.before(2, lambda: print('Limited access'))
Expected output: Function that only prints for the first 2 calls
Source Code
def before(n, func):
result = [None]
calls = [0]
def limited_func(*args, **kwargs):
if calls[0] < n:
result[0] = func(*args, **kwargs)
calls[0] += 1
return result[0]
return limited_func
wrap()
Wraps a function with another function
Implementation
Creates a new function that applies the wrapper to the original
Example
wrapped = UniCoreFW.wrap(lambda x: x + 1, lambda f, x: f(x) * 2)
Expected output: Function that adds 1 to input then multiplies by 2
Source Code
def wrap(func, wrapper):
def wrapped(*args, **kwargs):
return wrapper(func, *args, **kwargs)
return wrapped
negate()
Creates a function that negates a predicate
Implementation
Returns a new function that returns the logical NOT of the original
Example
is_odd = UniCoreFW.negate(lambda x: x % 2 == 0)
Expected output: Function that returns True for odd numbers
Source Code
def negate(func):
return lambda *args, **kwargs: not func(*args, **kwargs)
property_func()
Return a function that retrieves a property value by name.
Implementation
Args: prop_name: The property name to retrieve
Returns:
A function that gets the property from objects
Example
obj = {'a': 1, 'b': 2}
property_func('a')(obj)
Expected output: 1
Source Code
def property_func(prop_name: str) -> Callable:
return (
lambda obj: obj.get(prop_name)
if isinstance(obj, dict)
else getattr(obj, prop_name, None)
)
compose()
Composes multiple functions
Implementation
Creates a function that applies each function in reverse order
Example
composed = UniCoreFW.compose(lambda x: x + 1, lambda x: x * 2)
Expected output: Function that doubles the input then adds 1
Source Code
def compose(*funcs):
def composed(value):
for func in reversed(funcs):
value = func(value)
return value
return composed
delay()
Delays a function execution
Implementation
Uses threading.Timer to execute after a delay
Example
UniCoreFW.delay(lambda: print('Delayed'), 1000)
Expected output: Prints 'Delayed' after 1 second
Source Code
def delay(func, wait, *args, **kwargs):
threading.Timer(wait / 1000, func, args=args, kwargs=kwargs).start()
once_()
Creates a function that is restricted to execute func once.
Subsequent calls return the value of the first invocation.
Implementation
Args: func: The function to execute.
Returns:
A new function that wraps `func`, ensuring it is only called once.
Example
def greet():
return "Hello!"
greet_once = once_(greet)
greet_once() # "Hello!"
greet_once() # "Hello!"
Expected output: ``
Source Code
def once_(func):
called = False
result = None
def wrapper(*args, **kwargs):
nonlocal called, result
if not called:
result = func(*args, **kwargs)
called = True
return result
return wrapper
anyof()
Returns a function that takes a single argument and returns True if any of the given functions, when called with that argument, return True.
Implementation
Parameters: *funcs: Callable[[Any], bool] Zero or more predicate functions to test.
Returns:
Callable[[Any], bool]
A function that takes an argument and returns True if any of the given
predicates return True, False otherwise.
Example
def is_even(x):
return x % 2 == 0
def is_positive(x):
return x > 0
is_even_or_positive = anyof(is_even, is_positive)
is_even_or_positive(2) # True
is_even_or_positive(-3) # False
Expected output: ``
Source Code
def anyof(*funcs: Callable[[Any], bool]) -> Callable[[Any], bool]:
def wrapper(arg: Any) -> bool:
return any(f(arg) for f in funcs)
wrapper.__name__ = "anyof"
wrapper._argcount = 1 # type: ignore
return wrapper
allany()
Returns a function that checks if any of the given predicates return True for every item in the given list.
Implementation
Parameters: *preds: Callable[[Any], bool] Zero or more predicate functions to test.
Returns:
Callable[[List[Any]], bool]
A function that takes a list and returns True if every item in the list
satisfies at least one of the given predicates, False otherwise.
Example
def is_even(x):
return x % 2 == 0
def is_positive(x):
return x > 0
is_even_and_positive = allany(is_even, is_positive)
is_even_and_positive([2, 4, 6]) # True
is_even_and_positive([1, 3, 5]) # False
Expected output: ``
Source Code
def allany(*preds: Callable[[Any], bool]) -> Callable[[List[Any]], bool]:
def tester(arr: List[Any]) -> bool:
for x in arr:
if not any(p(x) for p in preds):
return False
return True
tester.__name__ = "allany"
tester.__doc__ = allany.__doc__
return tester
defer()
Defers a function execution
Implementation
Uses a minimal timer to execute after the current stack
Example
UniCoreFW.defer(lambda: print('Deferred'))
Expected output: Prints 'Deferred' after current execution completes
Source Code
def defer(func, *args, **kwargs):
threading.Timer(0, func, args=args, kwargs=kwargs).start()
bind()
Binds a function to arguments
Implementation
Creates a function with pre-filled arguments
Example
bound = UniCoreFW.bind(lambda x, y: x + y, None, 5)
Expected output: Function that adds 5 to its argument
Source Code
def bind(func, context, *args):
def bound_func(*extra_args):
return func(
*(args + extra_args)
) # Only pass args and extra_args without context
return bound_func
bind_all()
Binds specified methods of an object to the object itself
Implementation
Uses get to create bound methods
Example
obj = SomeClass()
UniCoreFW.bind_all(obj, 'method1', 'method2')
Expected output: Object with bound methods
Source Code
def bind_all(obj, *methodNames):
for method_name in methodNames:
if hasattr(obj, method_name):
bound_method = getattr(obj, method_name, None).__get__(obj)
setattr(obj, method_name, bound_method)
mixin()
Adds properties of an object as functions on UniCoreFW
Implementation
Dynamically adds methods to the UniCoreFW class
Example
UniCoreFW.mixin({'customFunc': lambda x: x * 3})
Expected output: Adds customFunc to UniCoreFW
Source Code
def mixin(obj):
for key, func in obj.items():
if callable(func):
setattr(UniCoreFW, key, func)
matches()
Returns a function that checks if an object matches key-value pairs
Implementation
Creates a predicate function for object matching
Example
is_admin = UniCoreFW.matches({'role': 'admin'})
result = is_admin({'name': 'John', 'role': 'admin'})
Expected output: True
Source Code
def matches(attrs):
def match(obj):
return all(obj.get(k) == v for k, v in attrs.items())
return match
invoke()
Calls a method on each item in an array
Implementation
Uses getattr to find and call methods
Example
results = UniCoreFW.invoke(['hello', 'world'], 'upper')
Expected output: ['HELLO', 'WORLD']
Source Code
def invoke(array, func_name, *args):
return [
getattr(item, func_name, None)(*args) if hasattr(item, func_name) else None
for item in array
]
_()
Factory method for creating a UniCoreFW instance
Implementation
Class method that returns a new instance
Example
_ = UniCoreFW._(collection)
result = _.filter(lambda x: x > 2).value()
Expected output: UniCoreFW instance wrapping filtered collection
Source Code
@classmethod
def _(cls, collection):
return cls(collection)