from functools import reduce, partial
from operator import itemgetter, attrgetter
from typing import Callable, TypeVar, Iterable
T = TypeVar('T')
R = TypeVar('R')
def compose(*functions: Callable[[Any], Any]) -> Callable[[Any], Any]:
return reduce(lambda f, g: lambda x: f(g(x)), functions)
def curry(func: Callable[..., R]) -> Callable[..., R]:
def curried(*args: Any) -> Any:
if len(args) >= func.__code__.co_argcount:
return func(*args)
return lambda *more: curried(*(args + more))
return curried
def memoize(func: Callable[[T], R]) -> Callable[[T], R]:
cache: dict[T, R] = {}
def memoized(arg: T) -> R:
if arg not in cache:
cache[arg] = func(arg)
return cache[arg]
return memoized
def pipeline(*funcs: Callable[[Any], Any]) -> Callable[[Any], Any]:
def piped(value: Any) -> Any:
for func in funcs:
value = func(value)
return value
return piped
def partition(predicate: Callable[[T], bool], items: Iterable[T]) -> tuple[list[T], list[T]]:
true_vals, false_vals = [], []
for item in items:
if predicate(item):
true_vals.append(item)
else:
false_vals.append(item)
return true_vals, false_vals
def group_by(key: Callable[[T], str], items: Iterable[T]) -> dict[str, list[T]]:
groups: dict[str, list[T]] = {}
for item in items:
k = key(item)
if k not in groups:
groups[k] = []
groups[k].append(item)
return groups
def take(n: int, items: Iterable[T]) -> list[T]:
result: list[T] = []
for i, item in enumerate(items):
if i >= n:
break
result.append(item)
return result
def drop(n: int, items: Iterable[T]) -> list[T]:
result: list[T] = []
for i, item in enumerate(items):
if i >= n:
result.append(item)
return result
def zip_with(func: Callable[[T, T], R],
items1: Iterable[T],
items2: Iterable[T]) -> list[R]:
return [func(a, b) for a, b in zip(items1, items2)]
def any_of(predicate: Callable[[T], bool], items: Iterable[T]) -> bool:
return any(predicate(item) for item in items)
def all_of(predicate: Callable[[T], bool], items: Iterable[T]) -> bool:
return all(predicate(item) for item in items)
def sorted_by(items: Iterable[T], key: Callable[[T], Any], reverse: bool = False) -> list[T]:
return sorted(items, key=key, reverse=reverse)
def find_first(predicate: Callable[[T], bool], items: Iterable[T]) -> T | None:
for item in items:
if predicate(item):
return item
return None
def chunk(size: int, items: list[T]) -> list[list[T]]:
return [items[i:i + size] for i in range(0, len(items), size)]
def intersperse(separator: T, items: list[T]) -> list[T]:
result: list[T] = []
for i, item in enumerate(items):
if i > 0:
result.append(separator)
result.append(item)
return result
@curry
def add(a: int, b: int, c: int) -> int:
return a + b + c
users = [
{"name": "Alice", "age": 30, "city": "NYC"},
{"name": "Bob", "age": 25, "city": "LA"},
{"name": "Carol", "age": 35, "city": "NYC"}
]
sorted_users = sorted_by(users, itemgetter("age"))
grouped = group_by(itemgetter("city"), users)