API
apply_utils
|
Apply input on function. |
|
Apply input on an async function. |
|
Returns the result of applying fn(*args). |
|
Invokes the specified method on an object with *args and **kwargs. |
|
Invokes the specified async method on an object with *args and **kwargs. |
|
Performs a call of the input function. |
currying
|
Make a function handle partial input, returning a function that expects the rest. |
data
|
Flattens a non homogeneous iterable. |
|
|
Display dataclass field as a tuple of JSON strings. |
|
|
Apply a function on a given value if it's not None. |
debug_utils
|
|
|
|
|
|
|
Replace regular compose calls with this to get a breakpoint at each step. |
|
Replace regular compose_left calls with this to get a breakpoint at each step. |
Debug exception in a pipeline stage by looking at the causal value. |
|
|
Logs given text and returns the execution input, can also log the input itself by using {} in the text string. |
|
|
|
dict_utils
|
Associate a key-value pair to the input dict. |
|
Turns a dictionary into a function from key to value or default if key is not there. |
|
Creates a function that returns coll[i0][i1]...[iX] where [i0, i1, ..., iX]==keys. |
|
get_in function, returning None if a key is not there. |
|
Returns coll[i0][i1]...[iX] where [i0, i1, ..., iX]==keys and None if a key is not there. |
|
get_in function, returning default if a key is not there. |
|
|
|
|
|
Retrieves a dictionary entry by its key attr. |
|
Retrieves a dictionary entry by its key attr. |
|
Retrieves a dictionary entry by its key attr. |
|
Builds an index with arbitrary amount of steps from an iterable. |
|
Given a dictionary, return a new dictionary with 'key' removed. |
|
Rename a key in a dictionary. |
excepts_decorator
|
|
|
Same as sync excepts only that the handler gets the original function params after the exception param. |
functional
|
Assert a function f on the input, printing the output of input_to_message(input) if assertion is False. |
|
Assert a function f on the input, printing the output of input_to_message(input) if assertion is False. |
|
Associate a value to the input dict given the path "keys". |
|
|
|
Generates elements from min to max. |
|
Concat two iterables. |
|
|
|
|
|
Return a new instance of the dataclass where new_dataclass_instance.attr_name = attr_transformer(dataclass_instance.attr_name) >>> @dataclasses.dataclass(frozen=True) . |
|
|
|
|
|
Drops the first n elements of a sequence. |
|
Drops the last n elements of a sequence. |
|
|
|
Returns a predicate that checks if an iterabel ends with another iterable. |
|
Check if two values are equal when applying f on both of them. |
|
Check if two strings are equal, ignoring case. |
|
Call the function call with the arguments flipped. |
|
Returns a unique identifier for the given function and input. |
|
|
Returns a unique identifier for the given function. |
|
|
|
|
Group a collection by a key function, when the value is given by a reducer function. |
|
|
|
Returns inner function ignoring the provided inputs. |
|
Introduces an element between each pair of elements in the input sequence. |
|
Intersects a group of collections. |
|
Raises the given exception. |
|
Stable id for function calls, can be used for caching. |
|
Returns a function that ignores the input and just raises the given exception. |
|
|
Partition all elements of sequence into tuples of length at most n. |
|
|
|
|
|
|
Add a value to the beginning of an iterable. |
|
|
|
Samples an iterable uniformly in one pass with O(k) memory. |
|
|
|
Skip the first n elements of a sequence. |
A sequence of overlapping subsequences. |
|
|
Return a new list containing all items from the iterable in ascending order >>> sort([5,2,4,1]) '[1,2,4,5]' |
|
Return a new list containing all items from the iterable in ascending order, sorted by a key. |
|
Return a new list containing all items from the iterable in descending order, sorted by a key. |
|
Return a new list containing all items from the iterable in descending order >>> sort([5,2,4,1]) '[5,4,2,1]' |
|
Add a value to the end of an iterable. |
|
Get an iterator for the first n elements of a sequence. |
|
Take elements from an iterable as long as elements pass some predicate. |
|
Return a JSON representation of a 'dictionary' or an object. |
|
Generates elements from max to min. |
|
A functional try/except block: if func fails with exc1, raise exc2. |
|
Return only unique elements of a sequence |
|
Return only unique elements of a sequence defined by function f |
|
Gets a (potentially nested) dictionary, key(s) and a function, and return new dictionary d' where d'[key] = func(d[key]). |
functional_async
|
Concat iterables of an async_generator. |
|
An async functional try/except block: await and return func on x. |
|
Runs given funcs serially until getting a succefull result. |
|
Returns an AsyncGenerator of the results after applying async f to each element of Iterable it |
|
Returns an AsyncGenerator of the results after applying f to each async element of it |
|
Runs a coroutine in a synchronous context, blocking until result arrives. |
functional_generic
|
Second-order composition of f1 over f2. |
|
Pass a value through a list of functions, return True iff all functions returned True-ish values. |
|
Map an iterable using a function, return True iff all mapped values are True-ish. |
|
Returns a list generated from the provided input. |
|
|
|
>>> f = anyjuxt(gamla.identity, gamla.greater_than(1), gamla.greater_than(10))
|
|
Map an iterable using a function, return True if at least one mapped value is True-ish. |
|
Returns a list generated from the provided input. |
|
Named transformations of a value using named functions. |
|
Second-order composition of f2 over f1. |
|
Serially run each function on tee'd copies of a sequence. |
|
Applies mappers to values according to predicates. |
|
|
|
Complement of a boolean function. |
|
Compose sync and async functions to operate in series. |
|
Compose sync and async functions to operate in series. |
|
Returns a function that applies an itterable of other functions into a single sink function. |
|
Like count_by_many but with a function that returns a single key. |
Counts elements of a collection by a key function f. |
|
|
Counts elements of a collection by a key function f. |
Constructs a function that filters elements of a given iterable for which function returns true. |
|
|
Constructs a function that maps elements of a given iterable using the given function. |
Constructs a function from a given higher order function and returns its first order counterpart. |
|
|
Turns a variadic function into an unary one that gets a dict of keywoded args to the original function. |
|
Constructs a function that will return the first element of an iterable, that returns True when used with the the given function. |
|
>>> f = find(gamla.greater_than(10))
|
|
Constructs a function that computes all functions from funcs, and returns the first function that doesn't throw an exception of type exception_type. |
|
Group a collection by a key function, when the value is given by a reducer function. |
|
Return a mapping `{y: {x s.t. |
|
|
|
Constructs a function that filters items of a given dictionary for which function returns true. |
|
Constructs a function that applies the given function to items of a given dictionary. |
|
Create a function that applies each function in :funcs: to its arguments and returns a tuple of the results. |
|
Create a function that calls the supplied functions, and chains the results. |
|
Create a function that filters a dict using a predicate over keys. |
|
Creates a function that maps supplied mapper over the keys of a dict. |
|
Create a function that applies each function in funcs to its arguments and returns a generator for the results. |
|
|
Constructs a function that maps elements of a given iterable using the given function. |
|
|
Constructs a function that maps elements of a given iterable using the given function. |
|
|
|
|
|
Returns a list generated from the provided input. |
|
Returns a function that given a value x, returns a tuple of the form: (x, f(x)). |
|
Returns a function that given a value x, returns a tuple of the form: (f(x), x). |
|
Pipe a value through a sequence of functions |
|
|
|
Constructs a function that removes elements of a given iterable for which function returns true. |
|
Like reduce, but keeps history of states. |
|
Runs f on x, returns x |
Note: Number of functions should be equal to the number of elements in the given iterable |
|
|
Turns a variadic function into an unary one that gets a tuple of args to the original function. |
|
Returns a function that computes f_true or f_false according to condition. |
|
Returns a function that computes f_false if condition is met. |
|
Create a function that filters a dict using a predicate over values. |
|
Creates a function then maps the supplied mapper over the values of a dict. |
|
Converts a string and Any input to a dict object. |
|
Returns f_true(args) if condition(args) returns true, else returns args. |
graph
|
|
|
|
|
Gets a directional graph and returns its sources. |
|
Gets a graph, a function to get a node's neighbours, a function to add to the set of seen nodes, and function to know if a node is seen or not. |
|
Gets a graph and return an iterator of its connectivity components. |
|
Gets a graph and a function to get a node's neighbours, BFS over it from a single source node, return an iterator of unique nodes. |
|
Gets a graph and a function to get a node's neighbours, BFS over it starting from multiple sources, return an iterator of unique nodes. |
|
Return a mapping `{y: {x s.t. y in f(x)}}, where x in it. `. |
|
Gets a graph, returns True if it contains a cycle, False else. |
|
Traverse over a graph like graph_traverse, but up to a given radius. |
graph_async
|
Gets a graph and a function to get a node's neighbours, BFS over it from a single source node, returns an AsyncGenerator of unique nodes. |
|
BFS over a graph, yielding unique nodes. |
|
Gets a graph and a function to get a node's neighbours, BFS over it from a single source node, returns an AsyncGenerator of unique nodes. |
|
Reduces a graph from some starting point using async functions. |
higher_order
|
|
|
|
Transforms a higher order function to a regular one. |
|
Transforms a higher order function to a regular one. |
io_utils
|
Batches single call into one request. |
|
Performs an async GET request to url with the specified timeout (seconds) and headers. |
|
Performs an async GET request to url with the specified timeout (seconds) and headers. |
|
Performs an async HEAD request to url with the specified timeout (seconds) and headers. |
|
Returns a function that can be used on any number of coroutines to make sure only limit amount of calls are done in parallel. |
|
Performs an http POST request of json data. |
Performs an http POST request with json data and URL parameters. |
|
Performs an http POST request of json data with specified headers. |
|
Queues identical calls to coroutine f and performs each call once. |
|
|
Creates a requests object. |
|
Wraps a coroutine to retry on given exceptions. |
|
Wraps a coroutine f assuring only limit amount of calls are done in parallel. |
|
Wraps a function f with a timer. |
|
Wraps a coroutine with a timeout (seconds) after which it will raise asyncio.TimeoutError. |
operator
|
Addition operator. |
|
Access the object attribute by its name attr. |
|
|
|
Contains operator. |
|
Counts the number of items in seq. |
|
|
|
|
|
|
Greater than or equal operator. |
|
|
Greater than operator. |
|
Returns the first element in a sequence. |
|
|
|
A functional in operator. |
|
Returns if the_value is an instance of the_type. |
|
Determines whether the element is iterable. |
|
Returns the last element in a sequence >>> last('ABC') 'C' |
|
Measures if the length of a sequence equals to a given length. |
|
Measures if the length of a sequence is greater than a given length. |
|
Measures if the length of a sequence is smaller than a given length. |
|
Less than or equal operator. |
|
Less than operator. |
|
Multiply operator. |
|
|
|
A functional !=. |
|
Returns the nth element in a sequence. |
|
Returns a list generated from the provided input. |
|
Returns the second element in a sequence. |
|
Returns the last n elements of a sequence. |
string_utils
|
Capitalize only the first letter of a given string, unlike str.capitalize that transforms all characters after the first character to lowercase. |
|
A curried regex match. |
|
Return a copy of the string with all occurrences of substring old replaced by new >>> txt = "hello world" >>> replace_in_text("world", "Jhon")(txt) 'hello Jhon' |
|
Return a list of the words in the string, using sep as the delimiter string |
tree
|
Gets a predicate, and builds a function that gets a dictionary, potentially nested and returns an iterable of leaf values. |
|
Gets a predicate, and builds a function that gets a dictionary, potentially nested and returns an iterable of leaf values. |
|
Reduce a JSON like tree. |
|
Like tree_reduce, but allows for an async map stage first, so it can be parallelized. |
|
Reduces a tree from the bottom up. |
|
Async version of tree_reduce. |
type_safety
|
Checks if destination can be composed after source, considering their typing. |
|
url_utils
|
Add params_to_add to the query string part of url |
Definitions
- gamla.apply_utils.apply_async(*args, **kwargs)[source]
Apply input on an async function.
>>> apply_async(1)(async_add(2)) 3
- gamla.apply_utils.apply_method(method: str, *args, **kwargs)[source]
Invokes the specified method on an object with *args and **kwargs.
>>> apply_method("get", "http://www.someurl.com")(requests) requests.Response()
- gamla.apply_utils.apply_method_async(method: str, *args, **kwargs)[source]
Invokes the specified async method on an object with *args and **kwargs.
>>> apply_method_async("get", "http://www.someurl.com")(httpx) httpx.Response()
- gamla.currying.curry(f)[source]
Make a function handle partial input, returning a function that expects the rest.
Warning: uses inspect which is slow, so avoid calling this inside a loop.
>>> def addition(a, b): return a + b ... add_3 = gamla.curry(addition)(3) ... add_3(7) 10
Can also be used as a decorator: >>>@gamla.curry … def addition(a, b): … return a + b
In case the function is async, the function becomes synchronous until the last argument. Although this is not always what you want, it fits the majority of cases.
>>> @gamla.curry ... async def async_addition(a, b): ... await asyncio.sleep(0.1) ... return a + b
>>> add_3 = async_addition(3)
>>> await add_3(7) # Note that `await` is needed here. Must be done inside an async scope.
The variables can be given with keywords, but mixing keyword and call by order might have unexpected results.
- gamla.data.explode(*positions: Collection[int])[source]
Flattens a non homogeneous iterable.
For an iterable where some positions are iterable and some are not, “explodes” the iterable, so that each element appears in a single row, and duplicates the non iterable.
>>> functional_generic.pipe( ... ["x", ["y1", "y2", "y3"], "z"], ... data.explode(1), ... tuple, ... ) ( ("x", "y1", "z"), ("x", "y2", "z"), ("x", "y3", "z"), )
- gamla.data.freeze_deep(*args, **kwargs)
Freeze recursively a dictionary.
>>> freeze_deep({"1": {"2": "345", "some-string": ["hello"]}}) data.frozendict( {"1": data.frozendict({"2": "345", "some-string": ("hello",)})}, )
- class gamla.data.frozendict(*args, **kwargs)[source]
- clear() None. Remove all items from D.
- pop(k[, d]) v, remove specified key and return the corresponding value.
If key is not found, d is returned if given, otherwise KeyError is raised
- popitem(*args, **kws)
Remove and return a (key, value) pair as a 2-tuple.
Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.
- setdefault(*args, **kws)
Insert key with a value of default if key is not in the dictionary.
Return the value for key if key is in the dictionary, else default.
- update([E, ]**F) None. Update D from dict/iterable E and F.
If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
- gamla.data.transform_if_not_none(transform: Callable, value)[source]
Apply a function on a given value if it’s not None. Else, return the None value.
>>> transform_if_not_none( ... functional_generic.when(operator.is_instance, lambda x: x.casefold())), ... "Some Title" ... ) 'some title' >>> transform_if_not_none( ... functional_generic.when(operator.is_instance, lambda x: x.casefold())), ... None ... )
- gamla.debug_utils.debug(*args, **kwargs)
A util to inspect a pipline by opening a debug prompt. Note: - Materializes generators, as most of the time we are interested in looking into them, so can have unexpected results. - The current value can be referenced by x in the debug prompt.
- gamla.debug_utils.debug_compose(*funcs)
Replace regular compose calls with this to get a breakpoint at each step.
- gamla.debug_utils.debug_compose_left(*funcs)
Replace regular compose_left calls with this to get a breakpoint at each step.
- gamla.debug_utils.debug_exception(f)[source]
Debug exception in a pipeline stage by looking at the causal value.
>>> gamla.pipe( ... "abc", ... gamla.itemgetter("some_key"), # This would cause an exception. ... )
>>> gamla.pipe( ... "abc", ... gamla.debug_exception(gamla.itemgetter("some_key")), # Now we can see the cause of the exception - we expect a `dict` but get a `str`. ... )
- gamla.debug_utils.log_text(text: str, level: int = 20)[source]
Logs given text and returns the execution input, can also log the input itself by using {} in the text string.
>>> f = log_text("hello world {}") >>> f("It's me!") INFO hello world It's me! 'It's me!'
- gamla.dict_utils.add_key_value(key, value)[source]
Associate a key-value pair to the input dict.
>>> add_key_value("1", "1")({"2": "2"}) {'2': '2', '1': '1'}
- gamla.dict_utils.dict_to_getter_with_default(default, d: Dict)[source]
Turns a dictionary into a function from key to value or default if key is not there.
>>> dict_to_getter_with_default(None, {1:1})(1) 1 >>> dict_to_getter_with_default(None, {1:1})(2) None
- gamla.dict_utils.get_in(keys: Iterable)[source]
Creates a function that returns coll[i0][i1]…[iX] where [i0, i1, …, iX]==keys.
>>> get_in(["a", "b", 1])({"a": {"b": [0, 1, 2]}}) 1
- gamla.dict_utils.get_in_or_none(keys: Iterable)[source]
get_in function, returning None if a key is not there.
>>> get_in_or_none(["a", "b", 1])({"a": {"b": [0, 1, 2]}}) 1 >>> get_in_or_none(["a", "c", 1])({"a": {"b": [0, 1, 2]}}) None
- gamla.dict_utils.get_in_or_none_uncurried(keys: Iterable, coll)[source]
Returns coll[i0][i1]…[iX] where [i0, i1, …, iX]==keys and None if a key is not there.
>>> get_in_or_none_uncurried(["a", "b", 1],{"a": {"b": [0, 1, 2]}}) 1 >>> get_in_or_none_uncurried(["a", "c", 1], {"a": {"b": [0, 1, 2]}}) None
- gamla.dict_utils.get_in_with_default(keys: Iterable, default)[source]
get_in function, returning default if a key is not there.
>>> get_in_with_default(["a", "b", 1], 0)({"a": {"b": [0, 1, 2]}}) 1 >>> get_in_with_default(["a", "c", 1], 0)({"a": {"b": [0, 1, 2]}}) 0
- gamla.dict_utils.itemgetter(attr)[source]
Retrieves a dictionary entry by its key attr.
>>> itemgetter("a")({"a": 1}) 1
- gamla.dict_utils.itemgetter_or_none(attr)[source]
Retrieves a dictionary entry by its key attr. Returns None if the key is not there.
>>> itemgetter_or_none("a")({"a": 1}) 1
>>> itemgetter_or_none("b")({"a": 1}) None
- gamla.dict_utils.itemgetter_with_default(default, attr)[source]
Retrieves a dictionary entry by its key attr. Returns default if the key is not there.
>>> itemgetter_with_default(0, "a")({"a": 1}) 1 >>> itemgetter_with_default(0, "b")({"a": 1}) 0
- gamla.dict_utils.make_index(steps: Iterable[Callable[[Iterable], Dict]]) Callable[[Iterable], Any][source]
Builds an index with arbitrary amount of steps from an iterable.
>>> index = dict_utils.make_index(map(gamla.groupby, [gamla.head, gamla.second]))(["uri", "dani"]) >>> index("d")("a") frozenset(["dani"])
- gamla.dict_utils.remove_key(key)[source]
Given a dictionary, return a new dictionary with ‘key’ removed. >>> remove_key(“two”)({“one”: 1, “two”: 2, “three”: 3}) {‘one’: 1, ‘three’: 3}
- gamla.dict_utils.rename_key(old: str, new: str) Callable[[dict], dict][source]
Rename a key in a dictionary.
>>> my_dict = {"name": "Danny", "age": 20} >>> rename_key("name", "first_name")(my_dict) {'first_name': 'Danny', 'age': 20}
- gamla.dict_utils.transform_item(key, f: Callable) Callable[[dict], dict][source]
transform a value of key in a dict. i.e given a dict d, return a new dictionary e s.t e[key] = f(d[key]).
>>> my_dict = {"name": "Danny", "age": 20} >>> transform_item("name", str.upper)(my_dict) {'name': 'DANNY', 'age': 20}
- gamla.excepts_decorator.try_and_excepts(exception: Tuple[Exception, ...] | Exception, handler: Callable, function: Callable)[source]
Same as sync excepts only that the handler gets the original function params after the exception param.
- gamla.functional.assert_that(f: Callable)
Assert a function f on the input, printing the output of input_to_message(input) if assertion is False.
>>> assert_that_with_message(just("Input is not 2!"), equals(2))(2) 2 >>> assert_that_with_message(just("Input is not 2!"), equals(2))(3) "Output is not 2!"
- gamla.functional.assert_that_with_message(input_to_message: Callable, f: Callable)[source]
Assert a function f on the input, printing the output of input_to_message(input) if assertion is False.
>>> assert_that_with_message(just("Input is not 2!"), equals(2))(2) 2 >>> assert_that_with_message(just("Input is not 2!"), equals(2))(3) "Output is not 2!"
- gamla.functional.assoc_in(d, keys, value, factory=<class 'dict'>)[source]
Associate a value to the input dict given the path “keys”.
>>> assoc_in({"a": {"b": 1}}, ["a", "b"], 2) {'a': {'b': 2}}
- gamla.functional.attr_equals(attribute: str, equals_what: Any) Callable[[Any], bool][source]
Returns a function that get an object x and returns whether x.attribute == equals_what
>>> attr_equals("imag", 5.0)(8 + 5j) True >>> attr_equals("imag", 5.0)(8) False
- gamla.functional.average(*args, **kwargs)
Average of an iterable. If the sequence is empty, returns 0. >>> average([1,2,3]) 2.0
- gamla.functional.bottom(iterable, key=<function identity>)[source]
Generates elements from min to max.
>>> tuple(bottom((3, 2, 1))) (1, 2, 3)
>>> tuple(bottom((1, 2, 3, 4), lambda x: x % 2 == 0)) (1, 3, 2, 4)
- gamla.functional.concat_with(new_it: Iterable, it: Iterable)[source]
Concat two iterables.
>>> tuple(concat_with((3, 4), (1, 2))) (1, 2, 3, 4)
- gamla.functional.dataclass_transform(attr_name: str, attr_transformer: Callable[[Any], Any])[source]
Return a new instance of the dataclass where new_dataclass_instance.attr_name = attr_transformer(dataclass_instance.attr_name) >>> @dataclasses.dataclass(frozen=True) … class C: … x: int >>> c = C(5) >>> d = dataclass_transform(‘x’, lambda i: i * 2)(c) >>> assert d.x == 10
- gamla.functional.drop(n: int)[source]
Drops the first n elements of a sequence.
>>> tuple(drop(2)([1,2,3,4,5])) (3,4,5)
- gamla.functional.drop_last(n: int)[source]
Drops the last n elements of a sequence.
>>> tuple(drop_last(1)([1,2,3,4,5])) (1,2,3,4)
- gamla.functional.duplicated_values(seq)[source]
Returns duplicated values from a given sequence.
>>> duplicated_values([1,2,2,3,3,4,5]) (2, 3)
- gamla.functional.ends_with(expected_tail: Iterable) Callable[[Sequence], bool][source]
Returns a predicate that checks if an iterabel ends with another iterable.
>>> ends_with([1,2,3])((0,1,2,3)) True >>> ends_with([1,2,3])((1,2)) False >>> ends_with([1,2])((3,1,2)) True >>> ends_with([1])(()) False
- gamla.functional.eq_by(f, value_1, value_2)[source]
Check if two values are equal when applying f on both of them.
- gamla.functional.eq_str_ignore_case(value_1, value_2)
Check if two strings are equal, ignoring case. >>> eq_str_ignore_case(“HeLlO wOrLd”, “hello world”) True
- gamla.functional.flip(func: Callable)[source]
Call the function call with the arguments flipped.
>>> import operator; flip(operator.truediv)(2, 6) 3.0
- gamla.functional.function_and_input_to_identifier(factory) Callable[source]
Returns a unique identifier for the given function and input.
- gamla.functional.function_to_directory(*args, **kwargs)
Directory path of a given function
- gamla.functional.function_to_uid(f: Callable) str[source]
Returns a unique identifier for the given function.
- gamla.functional.groupby_many_reduce(key: Callable, reducer: Callable, seq: Iterable)[source]
Group a collection by a key function, when the value is given by a reducer function.
Parameters: key (Callable): Key function (given object in collection outputs key). reducer (Callable): Reducer function (given object in collection outputs new value). seq (Iterable): Collection.
Returns: Dict[Text, Any]: Dictionary where key has been computed by the key function and value by the reducer function.
>>> groupby_many_reduce(head, lambda x, y: x + len(y) if x else len(y), ["hello", "hi", "test", "to"]) {'h': 7, 't': 6}
- gamla.functional.ignore_input(inner: Callable[[], Any]) Callable[source]
Returns inner function ignoring the provided inputs.
>>> ignore_input(lambda: 0)(1) 0
- gamla.functional.interpose(el)[source]
Introduces an element between each pair of elements in the input sequence.
>>> tuple(interpose("a")([1, 2, 3])) (1, 'a', 2, 'a', 3)
- gamla.functional.intersect(collections: Collection[Collection]) Iterable[source]
Intersects a group of collections.
Caller is responsible to give collections with O(1) containment checks.
>>> tuple( gamla.intersect( [ [1, 2, 3, 4], [4, 5], [4, 2], ] ) ) (4,)
- gamla.functional.just_raise(exception)[source]
Raises the given exception.
>>> just_raise(KeyError) raise exception KeyError
- gamla.functional.make_call_key(args, kwargs)[source]
Stable id for function calls, can be used for caching.
- gamla.functional.make_raise(exception)[source]
Returns a function that ignores the input and just raises the given exception.
>>> f = make_raise(KeyError) >>> f(3) raise exception KeyError
- gamla.functional.partition_all(n: int)[source]
Partition all elements of sequence into tuples of length at most n. The final tuple may be shorter to accommodate extra elements.
>>> list(partition_all(2)([1, 2, 3, 4])) [(1, 2), (3, 4)]
>>> list(partition_all(2)([1, 2, 3, 4, 5])) [(1, 2), (3, 4), (5,)]
- gamla.functional.prefix(val: Any, it: Iterable)[source]
Add a value to the beginning of an iterable. Return an iterable.
>>> tuple(prefix(1, (2, 3, 4))) (1, 2, 3, 4)
- gamla.functional.sample(k: int)
Samples an iterable uniformly in one pass with O(k) memory.
>>> sample(2)([1, 2, 3]) frozenset([1,3])
- gamla.functional.sample_with_randint(randint: Callable, k: int)[source]
Samples an iterable uniformly in one pass with O(k) memory.
>>> sample(2)([1, 2, 3]) frozenset([1,3])
- gamla.functional.skip(n: int)[source]
Skip the first n elements of a sequence. i.e, Return a generator that yields all elements after the n’th element. >>> tuple(skip(3)([i for i in range(6)])) (3, 4, 5)
- gamla.functional.sliding_window(n: int)[source]
A sequence of overlapping subsequences.
>>> list(sliding_window(2)([1, 2, 3, 4])) [(1, 2), (2, 3), (3, 4)]
- gamla.functional.sort(seq: Iterable)
Return a new list containing all items from the iterable in ascending order >>> sort([5,2,4,1]) ‘[1,2,4,5]’
- gamla.functional.sort_by(key: Callable)[source]
Return a new list containing all items from the iterable in ascending order, sorted by a key. >>> sort_by(len)([“hi!”, “my”, “name”, “is”]) [‘my’, ‘is’, ‘hi!’, ‘name’]
- gamla.functional.sort_by_reversed(key: Callable)[source]
Return a new list containing all items from the iterable in descending order, sorted by a key. >>> sort_by_reversed(lambda x: x % 10)([2231, 47, 19, 100]) [19, 47, 2231, 100]
- gamla.functional.sort_reversed(seq: Iterable)
Return a new list containing all items from the iterable in descending order >>> sort([5,2,4,1]) ‘[5,4,2,1]’
- gamla.functional.suffix(val: Any, it: Iterable)[source]
Add a value to the end of an iterable. Return an iterable.
>>> tuple(suffix(4, (1, 2, 3))) (1, 2, 3, 4)
- gamla.functional.take(n: int)[source]
Get an iterator for the first n elements of a sequence.
>>> tuple(take(3) ([1, 2, 3, 4, 5])) (1, 2, 3)
- gamla.functional.take_while(predicate: Callable[[Any], bool])[source]
Take elements from an iterable as long as elements pass some predicate.
>>> list(functional.take_while(lambda x: x < 7)([1, 2, 9, 2])) [1, 2]
- gamla.functional.to_json(obj)[source]
Return a JSON representation of a ‘dictionary’ or an object.
>>> to_json({"one": 1, "two": 2}) '{"one": 1, "two": 2}'
- gamla.functional.top(iterable: ~typing.Iterable, key=<function identity>)[source]
Generates elements from max to min.
>>> tuple(top((1, 3, 2))) (3, 2, 1)
>>> tuple(top(('a', 'aa', 'aaa'), len)) ('aaa', 'aa', 'a')
- gamla.functional.translate_exception(func: Callable, exc1: Exception, exc2: Exception)[source]
A functional try/except block: if func fails with exc1, raise exc2.
>>> from gamla import functional_generic >>> functional_generic.pipe(iter([]), translate_exception(next, StopIteration, ValueError)) ValueError Note: `func` is assumed to be unary.
- gamla.functional.unique(seq)
Return only unique elements of a sequence
>>> tuple(unique(["cat", "mouse", "dog", "cat"])) ('cat', 'mouse', 'dog')
- gamla.functional.unique_by(f)[source]
Return only unique elements of a sequence defined by function f
>>> tuple(unique_by(len)(['cat', 'mouse', 'dog', 'hen'])) ('cat', 'mouse')
- gamla.functional.update_in(d: dict, keys: ~typing.Iterable, func: ~typing.Callable, default=None, factory=<class 'dict'>)[source]
Gets a (potentially nested) dictionary, key(s) and a function, and return new dictionary d’ where d’[key] = func(d[key]).
>>> inc = lambda x: x + 1 >>> update_in({'a': 0}, ['a'], inc) {'a': 1}
- async gamla.functional_async.aconcat(async_generator: AsyncGenerator) AsyncGenerator[source]
Concat iterables of an async_generator.
>>> async def many_range(count): ... for i in range(count): ... yield range(i, i+1) ... async def to_list(ag): ... return [i async for i in ag] ... run_sync(to_list(aconcat(many_range(4)))) [0, 1, 2, 3]
- gamla.functional_async.aexcepts(exception_type: Type[Exception], func: Callable, handler: Callable, x: Any)[source]
An async functional try/except block: await and return func on x. If fails with exception_type, return the reult of running handler on the error.
>>> async def araise(x): ... raise ValueError ... run_sync(aexcepts(ValueError, araise, lambda e: e, 5)) ValueError()
>>> async def a_just(x): ... return x ... run_sync(aexcepts(ValueError, a_just, lambda e: e, 5)) 5
- gamla.functional_async.afirst(*funcs: Callable, exception_type: Type[Exception])[source]
Runs given funcs serially until getting a succefull result. Returns the result of the first function that runs on x without raising exception_type. If all given function raise e`xception_type`, exception_type will be raised.
>>> async def araise(x): ... raise ValueError ... run_sync(afirst(araise, lambda x: x*x, exception_type=ValueError)(3)) 9
- gamla.functional_async.amap_ascompleted(f: Callable[[Any], Awaitable[Any]], it: Iterable) AsyncGenerator[Any, None][source]
Returns an AsyncGenerator of the results after applying async f to each element of Iterable it
>>> async def amulti(x) ... return x*2 ... async def to_list(ag): ... return [i async for i in ag] ... run_sync(to_list(amap_ascompleted(amulti, range(4)))) [6, 0, 2, 4] (In a random order)
- gamla.functional_async.mapa(f: Callable, it: AsyncGenerator) AsyncGenerator[source]
Returns an AsyncGenerator of the results after applying f to each async element of it
>>> async def arange(count): ... for i in range(count): ... yield(i) ... async def to_list(ag): ... return [i async for i in ag] ... run_sync(to_list(mapa(lambda x: x*2, arange(4)))) [0, 2, 4, 6]
- gamla.functional_async.run_sync(f)[source]
Runs a coroutine in a synchronous context, blocking until result arrives.
>>> async def afoo(x): ... await asyncio.sleep(1) ... return x ... run_sync(afoo(1)) 1 (after 1 second of waiting)
- gamla.functional_generic.after(f1)[source]
Second-order composition of f1 over f2.
A ‘delayed’ pipeline, i.e return a function that, given f2, will wait for f2’s arguments, and when given, will return f1(f2(args)). >>> allmap = gamla.compose_left(gamla.map, gamla.after(all)) >>> allmap(lambda x: x == 1)([1, 2, 3]) False
- gamla.functional_generic.alljuxt(*funcs: Tuple[Callable, ...]) Callable[[...], Generator] | Callable[[...], Coroutine[None, None, Tuple]]
Pass a value through a list of functions, return True iff all functions returned True-ish values.
>>> f = alljuxt(gamla.identity, gamla.greater_than(1), gamla.greater_than(10)) >>> f(100) True >>> f(10) False
- gamla.functional_generic.allmap(f)
Map an iterable using a function, return True iff all mapped values are True-ish. >>> f = allmap(lambda x: x % 2 == 0) >>> f([1, 2, 3, 4, 5]) False >>> f([2, 4, 6, 8, 10]) True
- gamla.functional_generic.allstack(*stuff)
Runs packstack with given functions, then runs all on the output.
- gamla.functional_generic.anyjuxt(*funcs: Tuple[Callable, ...]) Callable[[...], Generator] | Callable[[...], Coroutine[None, None, Tuple]]
>>> f = anyjuxt(gamla.identity, gamla.greater_than(1), gamla.greater_than(10)) >>> f(100) True >>> f(10) True >>> f(0) False
- gamla.functional_generic.anymap(f)
Map an iterable using a function, return True if at least one mapped value is True-ish. Note: evaluation is lazy, i.e. returns on first True. >>> f = anymap(lambda x: x % 2 == 0) >>> f([1, 2, 3, 4, 5]) True >>> f([1, 3, 5, 7, 9]) False
- gamla.functional_generic.anystack(*stuff)
Runs packstack with given functions, then runs any on the output.
- gamla.functional_generic.apply_spec(spec: Dict)[source]
Named transformations of a value using named functions.
>>> spec = {"len": len, "sum": sum} >>> apply_spec(spec)([1,2,3,4,5]) {'len': 5, 'sum': 15}
Notes: - The dictionary can be nested. - Returned function will be async iff any leaf is an async function.
- gamla.functional_generic.bifurcate(*funcs)[source]
Serially run each function on tee’d copies of a sequence. If the sequence is a generator, it is duplicated so it will not be exhausted (which may incur a substantial memory signature in some cases). >>> f = bifurcate(sum, gamla.count) >>> seq = map(gamla.identity, [1, 2, 3, 4, 5]) >>> f(seq) (15, 5)
- gamla.functional_generic.case(predicates_and_mappers: Tuple[Tuple[Callable, Callable], ...])[source]
Applies mappers to values according to predicates. If no predicate matches, raises gamla.functional_generic.NoConditionMatched. >>> f = case(((gamla.less_than(10), gamla.identity), (gamla.greater_than(10), gamla.add(100)))) >>> f(5) 5 >>> f(15) 115 >>> f(10) NoConditionMatched
- gamla.functional_generic.case_dict(*args, **kwargs)
Applies functions to values according to predicates given in a dict. Raises gamla.functional_generic.NoConditionMatched if no predicate matches. >>> f = case_dict({gamla.less_than(10): gamla.identity, gamla.greater_than(10): gamla.add(100)}) >>> f(5) 5 >>> f(15) 115 >>> f(10) NoConditionMatched
- gamla.functional_generic.complement(f2)
Complement of a boolean function.
>>> f = complement(gamla.greater_than(5)) >>> f(10) False >>> f(1) True
- gamla.functional_generic.compose(*funcs)[source]
Compose sync and async functions to operate in series.
Returns a function that applies other functions in sequence. The returned function will be an async function iff at least one of the functions in the sequence is async.
Functions are applied from right to left so that
compose(f, g, h)(x, y)is the same asf(g(h(x, y))).>>> inc = lambda i: i + 1 >>> compose(str, inc)(3) '4'
- See Also:
compose_left pipe
- gamla.functional_generic.compose_left(*funcs)[source]
Compose sync and async functions to operate in series.
Returns a function that applies other functions in sequence. The returned function will be an async function iff at least one of the functions in the sequence is async.
Functions are applied from left to right so that
compose_left(f, g, h)(x, y)is the same ash(g(f(x, y))).>>> inc = lambda i: i + 1 >>> compose_left(inc, str)(3) '4'
- See Also:
compose pipe
- gamla.functional_generic.compose_many_to_one(incoming: Iterable[Callable], f: Callable)[source]
Returns a function that applies an itterable of other functions into a single sink function. The returned function will be an async function iff at least one of the given functions is async.
compose_many_to_one([f, g, k], h)(x, y)is the same ash(f(x,y), g(x, y), k(x, y)).>>> compose_many_to_one([sum, sum], lambda x, y: x + y)([1, 2, 3]) 12
- See Also:
juxt compose_left
- gamla.functional_generic.count_by(f2) Dict[Any, int]
Like count_by_many but with a function that returns a single key.
- gamla.functional_generic.count_by_many(f: Callable[[Any], Iterable]) Dict[Any, int][source]
Counts elements of a collection by a key function f.
>>> count_by_many( ... gamla.juxt( ... operator.head, ... gamla.last, ... ) ... )(["aa", "ab", "ac", "bc"]) {'a': 3, 'b': 2, 'c': 2}
- gamla.functional_generic.countby_many(f: Callable[[Any], Iterable]) Dict[Any, int]
Counts elements of a collection by a key function f.
>>> count_by_many( ... gamla.juxt( ... operator.head, ... gamla.last, ... ) ... )(["aa", "ab", "ac", "bc"]) {'a': 3, 'b': 2, 'c': 2}
- gamla.functional_generic.curried_filter(f)
Constructs a function that filters elements of a given iterable for which function returns true. Returns an async function iff the filter function is async, else returns a sync function.
>>> f = curried_filter(gamla.greater_than(10)) >>> f([1, 2, 3, 11, 12, 13]) [11, 12, 13]
- gamla.functional_generic.curried_map(f)[source]
Constructs a function that maps elements of a given iterable using the given function.
Returns an async function iff f is async, else returns a sync function.
>>> inc = lambda i: i + 1 >>> curried_map(inc)([3, 4, 5]) [4, 5, 6]
- gamla.functional_generic.curried_to_binary(f)[source]
Constructs a function from a given higher order function and returns its first order counterpart. The given higher order function, f is must be a unary function Returns an async function iff f is async, else returns a sync function.
>>> inc = lambda i: i + 1 >>> f = curried_to_binary(curried_map) >>> f(inc, [1, 2, 3]) [2, 3, 4]
- gamla.functional_generic.double_star(f)
Turns a variadic function into an unary one that gets a dict of keywoded args to the original function.
>>> pipe(({"x": 2, "y": 3}), double_star(lambda x, y: x + y)) 5
- gamla.functional_generic.find(f)
Constructs a function that will return the first element of an iterable, that returns True when used with the the given function. If no element in the iterable returns True, None is returned.
>>> f = find(gamla.greater_than(10)) >>> f([1, 2, 3, 11, 12, 13]) 11 >>> f([1, 2, 3]) None
- gamla.functional_generic.find_index(f2)
>>> f = find(gamla.greater_than(10)) >>> f([1, 2, 3, 11, 12, 13]) 11 >>> f([1, 2, 3]) -1
- See Also:
find
- gamla.functional_generic.first(*funcs, exception_type: Type[Exception])[source]
Constructs a function that computes all functions from funcs, and returns the first function that doesn’t throw an exception of type exception_type. The function is async if at least one of the given functions is async. If all functions throw the given exception_type, exception_type will be raised.
>>> f = gamla.first(gamla.second, gamla.head, exception_type=StopIteration) >>> f([1,2]) '2' >>> f([1]) '1' >>> f([]) StopIteration raised
- gamla.functional_generic.frequencies(seq: Iterable)
Return a dict with number of occurrences of each value in a sequence. >>> frequencies([‘cat’, ‘cat’, ‘ox’, ‘pig’, ‘pig’, ‘cat’]) {‘cat’: 3, ‘ox’: 1, ‘pig’: 2}
- gamla.functional_generic.groupby(key: Callable[[_ReducedElement], _K]) Callable[[Iterable[_ReducedElement]], Mapping[_K, Tuple[_ReducedElement, ...]]][source]
Return a mapping {y: {x s.t. key(x) = y}}.
>>> names = ['alice', 'bob', 'barbara', 'frank', 'fred'] >>> f = groupby(gamla.head) >>> f(names) {"a": ("alice",), "b": ("bob", "barbara"), "f": ("frank", "fred")}
- gamla.functional_generic.itemfilter(f)
Constructs a function that filters items of a given dictionary for which function returns true. Returns an async function iff the filter function is async, else returns a sync function.
>>> f = itemfilter( ... alljuxt( ... compose_left(gamla.head, gamla.contains("gamla")), ... compose_left(gamla.second, gamla.greater_than(10)), ... ) ... ) >>> f({"gamla": 11, "gaml": 9, "f":12}) {'gamla': 11}
- gamla.functional_generic.itemmap(f)
Constructs a function that applies the given function to items of a given dictionary. Returns an async function iff the filter function is async, else returns a sync function.
>>> f = itemmap(gamla.star(lambda key, val: (key, val + key))) >>> f({1: 2, 2: 3}) {1: 3, 2: 5}
- gamla.functional_generic.juxt(*funcs: Callable) Callable[[...], Tuple][source]
Create a function that applies each function in :funcs: to its arguments and returns a tuple of the results.
>>> inc = lambda x: x + 1 >>> double = lambda x: x * 2 >>> juxt(inc, double)(10) (11, 20)
- gamla.functional_generic.juxtcat(*funcs: Tuple[Callable, ...]) Callable[[...], Generator] | Callable[[...], Coroutine[None, None, Tuple]]
Create a function that calls the supplied functions, and chains the results. Assumes the supplied functions return Iterables. >>> f = juxtcat(range,range) >>> tuple(f(5)) (0 ,1 ,2 ,3 ,4 ,0 ,1 ,2 ,3 ,4)
- gamla.functional_generic.keyfilter(f2)
Create a function that filters a dict using a predicate over keys.
>>> f = keyfilter(lambda k: k > 2) >>> f({1:"a",2:"b",3:"c",4:"d"}) { 3: "c", 4: "d" }
- gamla.functional_generic.keymap(f2)
Creates a function that maps supplied mapper over the keys of a dict.
>>> f = keymap(lambda k: k + 1) >>> f({1:"a",2:"b",3:"c",4:"d"}) { 2: "a", 3: "b", 4: "c", 5: "d" }
- gamla.functional_generic.lazyjuxt(*funcs: Tuple[Callable, ...]) Callable[[...], Generator] | Callable[[...], Coroutine[None, None, Tuple]][source]
Create a function that applies each function in funcs to its arguments and returns a generator for the results.
Applies the supplied functions lazily as the returned generator is iterated. Reverts to eager implementation if any of funcs is async.
>>> inc = lambda x: x + 1 >>> double = lambda x: x * 2 >>> tuple(lazyjuxt(inc, double)(10)) (11, 20)
- gamla.functional_generic.map_filter_empty(f)
Constructs a function that maps elements of a given iterable using the given function.
Returns an async function iff f is async, else returns a sync function.
>>> inc = lambda i: i + 1 >>> curried_map(inc)([3, 4, 5]) [4, 5, 6]
- gamla.functional_generic.mapcat(f)
Constructs a function that maps elements of a given iterable using the given function.
Returns an async function iff f is async, else returns a sync function.
>>> inc = lambda i: i + 1 >>> curried_map(inc)([3, 4, 5]) [4, 5, 6]
- gamla.functional_generic.packstack(*stuff)
Like stack but doesn’t require additional brackets.
- gamla.functional_generic.pair_right(f)[source]
Returns a function that given a value x, returns a tuple of the form: (x, f(x)).
>>> add_one = pair_right(lambda x: x + 1) >>> add_one(3) (3, 4)
- gamla.functional_generic.pair_with(f)[source]
Returns a function that given a value x, returns a tuple of the form: (f(x), x).
>>> add_one = pair_with(lambda x: x + 1) >>> add_one(3) (4, 3)
- gamla.functional_generic.pipe(val, *funcs)[source]
Pipe a value through a sequence of functions
I.e.
pipe(val, f, g, h)is equivalent toh(g(f(val)))>>> double = lambda i: 2 * i >>> pipe(3, double, str) '6'
- gamla.functional_generic.remove(f2)
Constructs a function that removes elements of a given iterable for which function returns true. Returns an async function iff the filter function is async, else returns a sync function.
>>> f = remove(gamla.greater_than(10)) >>> tuple(f([1, 2, 3, 11, 12, 13])) (1, 2, 3)
- gamla.functional_generic.scan(reducer: Callable[[_ReducerState, _ReducedElement], _ReducerState], initial_value: _ReducerState) Callable[[Iterable[_ReducedElement]], Tuple[_ReducerState, ...]][source]
Like reduce, but keeps history of states.
See https://en.wikipedia.org/wiki/Prefix_sum#Scan_higher_order_function.
- gamla.functional_generic.side_effect(f: Callable)[source]
Runs f on x, returns x
>>> log_and_add = compose_left(side_effect(print), add(1))) >>> log_and_add(2) 2 3
- gamla.functional_generic.stack = <function enumerate>
Note: Number of functions should be equal to the number of elements in the given iterable
>>> stack([lambda x:x+1, lambda x:x-1])((5, 5)) (6, 4)
- gamla.functional_generic.star(f)
Turns a variadic function into an unary one that gets a tuple of args to the original function.
>>> pipe((2, 3), star(lambda x, y: x + y)) 5
- gamla.functional_generic.ternary(condition, f_true, f_false)[source]
Returns a function that computes f_true or f_false according to condition. The functions are applied with the same input. The returned function will be an async function if at least one of the given functions is async.
>>> f = ternary(gamla.greater_than(5), gamla.identity, lambda i: -i) >>> f(6) '6' >>> f(3) '-3'
- gamla.functional_generic.unless(condition, f_false)[source]
Returns a function that computes f_false if condition is met. Otherwise will return the input unchanged. condition and f_false are applied with the same input.
>>> f = unless(gamla.greater_than(5), lambda i: -i) >>> f(6) '6' >>> f(3) '-3'
- gamla.functional_generic.valfilter(f2)
Create a function that filters a dict using a predicate over values.
>>> f = valefilter(lambda k: k > 2) >>> f({"a": 1, "b": 2, "c": 3, "d": 4}) { "c": 3, "d": 4 }
- gamla.functional_generic.valmap(f2)
Creates a function then maps the supplied mapper over the values of a dict.
>>> f = valmap(gamla.add(1)) >>> f({ "a": 1, "b": 2, "c": 3 }) { "a": 2 "b": 3 "c": 4 }
- gamla.functional_generic.value_to_dict(key: str)[source]
Converts a string and Any input to a dict object.
>>> value_to_dict("hello")("world") {'hello': 'world'}
- gamla.functional_generic.when(condition: Callable, f_true: Callable) Callable[source]
Returns f_true(args) if condition(args) returns true, else returns args.
>>> f = when(gamla.greater_than(5), lambda i: -i) >>> f(6) '-6' >>> f(3) '3'
- gamla.graph.cliques_to_graph(it)
Gets a sequence of nodes (cliques) and returns the bidirectional graph they represent
>>> cliques_to_graph([{1, 2}, {3, 4}]) {1: frozenset({2}), 2: frozenset({1}), 3: frozenset({4}), 4: frozenset({3})}
- gamla.graph.edges_to_graph(it)
Gets a sequence of edges and returns a graph made of these edges.
>>> graph.edges_to_graph([(1,2), (2, 3), (3, 1), (3, 2)]) {1: (2,), 2: (3,), 3: (1, 2)}
- gamla.graph.find_sources(graph: Dict) FrozenSet[source]
Gets a directional graph and returns its sources.
>>> find_sources({'1': ['2', '3'], '2': ['3'], '3': [], '4': []}) frozenset({'1', '4'})
- gamla.graph.general_graph_traverse_many(sources: Any, get_neighbors: Callable, remember: Callable, should_traverse: Callable) Iterable[source]
Gets a graph, a function to get a node’s neighbours, a function to add to the set of seen nodes, and function to know if a node is seen or not. BFS over the graph and return an iterator of unique nodes.
>>> seen_set = set(); key = len; g = {'one': ['two', 'three'], 'two': ['three'], 'three': ['four'], 'four': []} >>> list(general_graph_traverse_many(['one', 'three'], g.__getitem__, functional_generic.compose_left(key, seen_set.add), functional_generic.compose_left(key, sync.complement(operator.contains(seen_set))))) ['three', 'one', 'four']
Note: get_neighbors must return elements without duplicates.
- gamla.graph.get_connectivity_components(graph: Dict) Iterable[FrozenSet][source]
Gets a graph and return an iterator of its connectivity components.
>>> g = cliques_to_graph([{1, 2}, {3, 4}]) >>> list(get_connectivity_components(g)) [frozenset({1, 2}), frozenset({3, 4})]
Note: Graph is assumed to undirected, so each edge must appear both ways.
- gamla.graph.graph_to_edges(d)
Gets a graph and returns an iterator of all edges in it.
>>> list(graph_to_edges({'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []})) [('1', '2'), ('1', '3'), ('2', '3'), ('3', '4')]
- gamla.graph.graph_traverse(source: ~typing.Any, get_neighbors: ~typing.Callable, key: ~typing.Callable = <function identity>) Iterable[source]
Gets a graph and a function to get a node’s neighbours, BFS over it from a single source node, return an iterator of unique nodes.
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> list(graph_traverse('1', g.__getitem__)) ['1', '2', '3', '4']
- gamla.graph.graph_traverse_many(sources: ~typing.Any, get_neighbors: ~typing.Callable, key: ~typing.Callable = <function identity>) Iterable[source]
Gets a graph and a function to get a node’s neighbours, BFS over it starting from multiple sources, return an iterator of unique nodes.
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> list(graph_traverse_many(['1', '3'], g.__getitem__)) ['3', '1', '4', '2']
Note: get_neighbors must return elements without duplicates.
- gamla.graph.groupby_many(f: Callable, it: Iterable) Dict[str, Any][source]
Return a mapping `{y: {x s.t. y in f(x)}}, where x in it. `
Parameters: Key function (gets an object in collection and outputs tuple of keys). A Collection.
Returns a dictionary where key has been computed by the f key function.
>>> names = ['alice', 'bob', 'charlie', 'dan', 'edith', 'frank'] >>> groupby_many(lambda name: (name[0], name[-1]), names) {'a': frozenset({'alice'}), 'e': frozenset({'alice', 'charlie', 'edith'}), 'b': frozenset({'bob'}), 'c': frozenset({'charlie'}), 'd': frozenset({'dan'}), 'n': frozenset({'dan'}), 'h': frozenset({'edith'}), 'f': frozenset({'frank'}), 'k': frozenset({'frank'})}
- gamla.graph.has_cycle(graph)[source]
Gets a graph, returns True if it contains a cycle, False else.
>>> has_cycle({1: [2], 2: [3], 3: [1]}) True >>> has_cycle({1: [2], 2: [3]}) False
- gamla.graph.reverse_graph(d)
Gets a graph and returns the graph with its edges reversed
>>> reverse_graph({'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []}) {'2': frozenset({'1'}), '3': frozenset({'1', '2'}), '4': frozenset({'3'})}
- gamla.graph.traverse_graph_by_radius(source: Any, get_neighbors: Callable, radius: int) Iterable[source]
Traverse over a graph like graph_traverse, but up to a given radius.
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> list(traverse_graph_by_radius('1', g.__getitem__, 1)) ['1', '2', '3']
- gamla.graph_async.agraph_traverse(source: ~typing.Any, aget_neighbors: ~typing.Callable[[~typing.Any], ~typing.AsyncGenerator], key: ~typing.Callable = <function identity>) AsyncGenerator[source]
Gets a graph and a function to get a node’s neighbours, BFS over it from a single source node, returns an AsyncGenerator of unique nodes.
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> async def get_item(x): >>> for a in g.get(x): >>> yield a >>> async def to_list(ag): >>> return [i async for i in ag] >>> gamla.run_sync(to_list(gamla.agraph_traverse('1', get_item))) ['1', '2', '3', '4']
- gamla.graph_async.agraph_traverse_many(sources: ~typing.Any, aget_neighbors: ~typing.Callable[[~typing.Any], ~typing.AsyncGenerator], key: ~typing.Callable = <function identity>) AsyncGenerator[Any, None][source]
BFS over a graph, yielding unique nodes. Use when aget_neighbors returns an AsyncGenerator.
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> async def get_item(x): >>> for a in g.get(x): >>> yield a >>> async def to_list(ag): >>> return [i async for i in ag] >>> gamla.run_sync(to_list(gamla.agraph_traverse_many(['1', '3'], get_item))) ['3', '1', '4', '2']
Note: aget_neighbors must return elements without duplicates.
- gamla.graph_async.atraverse_graph_by_radius(source: ~typing.Any, aget_neighbors: ~typing.Callable[[~typing.Any], ~typing.AsyncGenerator], radius: int, key: ~typing.Callable = <function identity>) AsyncGenerator[Any, None][source]
Gets a graph and a function to get a node’s neighbours, BFS over it from a single source node, returns an AsyncGenerator of unique nodes. Does not traverse farther from given radius
>>> g = {'1': ['2', '3'], '2': ['3'], '3': ['4'], '4': []} >>> async def get_item(x): >>> for a in g.get(x): >>> yield a >>> async def to_list(ag): >>> return [i async for i in ag] >>> gamla.run_sync(to_list(gamla.atraverse_graph_by_radius('1', get_item, 1))) ['1', '2', '3']
- gamla.graph_async.reduce_graph_async(reducer: Callable[[Tuple[Any, ...], Hashable], Any], get_neighbors: Callable, remember: Callable[[Hashable], None], is_seen: Callable[[Hashable], bool], current: Hashable)[source]
Reduces a graph from some starting point using async functions.
>>> set_instance = set() >>> await reduce_graph_async( ... lambda children, current: sum(children) + current, ... functional_generic.compose_left( ... dict_utils.dict_to_getter_with_default( ... (), ... {1: (1, 2, 3, 5), 2: (4,), 3: (1, 2)}), ... async_functions.to_awaitable, ... ), ... set_instance.add, ... contains(set_instance) ... 1, ... ) 15
- gamla.higher_order.on_first(f2)
Make a function act on the first element on incoming input.
- gamla.higher_order.on_second(f2)
Make a function act on the second element on incoming input.
- gamla.higher_order.persistent_cache(get_item: Callable[[str], Any], set_item: Callable[[str, Any], None], make_key: Callable[[Any], str]) Callable[source]
Wraps a function with persistent cache. Gets the item getter and item setter as parameters.
- gamla.higher_order.prepare_and_apply(f: Callable) Callable[source]
Transforms a higher order function to a regular one.
Uses the given value once to prepare a regular function, then again to call it with.
>>> def increment(x): return x + 1 >>> def decrement(x): return x - 1 >>> def conditional_transformation(x): ... return increment if x < 10 else decrement >>> prepare_and_apply(conditional_transformation))(15) 14
- gamla.higher_order.prepare_and_apply_async(f: Callable) Callable[source]
Transforms a higher order function to a regular one.
Uses the given value once to prepare a regular function, then again to call it with.
>>> async def increment(x): return x + 1 >>> async def decrement(x): return x - 1 >>> def conditional_transformation(x): ... return increment if x < 10 else decrement >>> prepare_and_apply(conditional_transformation))(15) 14
- gamla.io_utils.batch_calls(max_batch_size: int, f: Callable)[source]
Batches single call into one request. Turns f, a function that gets a tuple of independent requests, into a function that gets a single request. Each request will be at most of size max_batched_size. Requests will be performed in time intervals of 0.1s.
>>> batched_f = batch_calls(5, f)
- gamla.io_utils.get_async(timeout: float, url: str)
Performs an async GET request to url with the specified timeout (seconds) and headers. Expects the payload to be a json serializable object.
>>> response = await get_async(30, "https://www.someurl.com")
- gamla.io_utils.get_async_with_headers(headers: Dict[str, str], timeout: float, url: str)[source]
Performs an async GET request to url with the specified timeout (seconds) and headers.
>>> response = await get_async_with_headers({some_header: some_value}, 30, "https://www.someurl.com")
- gamla.io_utils.head_async_with_headers(headers: Dict[str, str], timeout: float, url: str)[source]
Performs an async HEAD request to url with the specified timeout (seconds) and headers.
>>> response = await head_async_with_headers(30, "https://www.someurl.com", {})
- gamla.io_utils.make_throttler(limit)[source]
Returns a function that can be used on any number of coroutines to make sure only limit amount of calls are done in parallel.
>>> throttler = make_throttler(3) >>> throttler(get_async) >>> throttler(post_json_async)
- gamla.io_utils.post_json_async(timeout: float, url: str, payload)
Performs an http POST request of json data. Expects the payload to be a json serializable object.
>>> response = post_json_async(30, "https://www.someurl.com/post_data", { "name": "Danny" })
- gamla.io_utils.post_json_with_extra_headers_and_params_async(params: Dict[str, str], extra_headers: Dict[str, str], timeout: float, url: str, payload)[source]
Performs an http POST request with json data and URL parameters. Additional headers may be specified. Expects the params to be a dictionary object and the payload to be a json serializable object.
>>> response = await post_json_with_extra_headers_and_params_async({"Date": "07/08/2022"}, {"Authorization": "Bearer TOKEN" }, 30, "https://www.someurl.com/post_data", { "name": "Danny" })
- gamla.io_utils.post_json_with_extra_headers_async(extra_headers: Dict[str, str], timeout: float, url: str, payload)
Performs an http POST request of json data with specified headers. Expects the payload to be a json serializable object and the headers to be a dictionary.
>>> response = post_json_with_extra_headers_async({"username": "First Last", 30, "https://www.someurl.com/post_data", { "name": "Danny" })
- gamla.io_utils.queue_identical_calls(f)[source]
Queues identical calls to coroutine f and performs each call once. Note that pending grows infinitely large, this assumes we cache the results anyway after this decorator, so we at most double the memory consumption.
>>> deduped_get_async = queue_identical_calls(get_async)
- gamla.io_utils.requests_with_retry(retries: int = 3) Session[source]
Creates a requests object. Request will be attempted retries times. Will retry on 500, 502 and 504 status codes.
>>> response = requests_with_retry(3).get("http://someurl.com")
- gamla.io_utils.retry(exception: Exception | Tuple[Exception, ...], times: int, wait_seconds: float, f: Callable)[source]
Wraps a coroutine to retry on given exceptions.
- gamla.io_utils.retry_with_count(exception: Exception | Tuple[Exception, ...], times: int, wait_seconds: float, f: Callable)[source]
Wraps a coroutine to retry on given exceptions and returns tuple of output and count of retries.
- gamla.io_utils.sleep(seconds: float) Callable[source]
Curried sleep to be used inside compositions.
>>> gamla.compose_left(fetch, gamla.side_effect(sleep(2)))
- gamla.io_utils.throttle(limit, f)[source]
Wraps a coroutine f assuring only limit amount of calls are done in parallel.
>>> throttled_get_async = throttle(3, get_async)
- gamla.io_utils.timeit(f)[source]
Wraps a function f with a timer. Logs the start time, and end time (and difference in seconds).
>>> timed_get_async = timeit(get_async)
- gamla.io_utils.timeout(seconds: float)[source]
Wraps a coroutine with a timeout (seconds) after which it will raise asyncio.TimeoutError.
>>> get_with_timeout = timeout(3.5)(get_async)
- gamla.operator.add(x)[source]
Addition operator.
>>> add(1)(2) 3
>>> add(["c"])(["a", "b"]) ['a', 'b', 'c']
- gamla.operator.attrgetter(attr)[source]
Access the object attribute by its name attr.
>>> attrgetter("lower")("ASD")() 'asd'
- gamla.operator.concat(iterable, /)
Alternative chain() constructor taking a single iterable argument that evaluates lazily.
- gamla.operator.contains(x)[source]
Contains operator.
>>> contains([1, 2, 3])(2) True
>>> contains("David")("x") False
- gamla.operator.count(seq: Iterable) int[source]
Counts the number of items in seq. Similar to len but works on lazy sequences.
- gamla.operator.greater_equals(x)[source]
Greater than or equal operator.
>>> greater_equals(1)(1) True
>>> greater_equals(1)(0) False
- gamla.operator.greater_than(x)[source]
Greater than operator.
>>> greater_than(1)(2) True
>>> greater_than(1)(0) False
- gamla.operator.inside(val)[source]
A functional in operator.
>>> inside(1)([0, 1, 2]) True
>>> inside("a", "word") False
- gamla.operator.is_instance(the_type)[source]
Returns if the_value is an instance of the_type.
>>> is_instance(str)("hello") True
>>> is_instance(int)("a") False
- gamla.operator.is_iterable(x)[source]
Determines whether the element is iterable.
>>> isiterable([1, 2, 3]) True >>> isiterable('abc') True >>> isiterable(5) False
- gamla.operator.len_equals(length: int)[source]
Measures if the length of a sequence equals to a given length.
>>> len_equals(3)([0, 1, 2]) True
- gamla.operator.len_greater(length: int)[source]
Measures if the length of a sequence is greater than a given length.
>>> len_greater(2)([0, 1, 2]) True
- gamla.operator.len_smaller(length: int) Callable[source]
Measures if the length of a sequence is smaller than a given length.
>>> len_smaller(2)([0, 1, 2]) False
- gamla.operator.less_equals(x)[source]
Less than or equal operator.
>>> less_equals(1)(1) True
>>> less_equals(1)(3) False
- gamla.operator.not_equals(x)[source]
A functional !=.
>>> not_equals(2)(2) False
>>> not_equals("David")("Michael") True
- gamla.operator.pack(*stuff)[source]
Returns a list generated from the provided input.
>>> pack(1, 2, 3) (1, 2, 3)
- gamla.operator.tail(n)[source]
Returns the last n elements of a sequence. >>> tail(2, [10, 20, 30, 40, 50]) [40, 50]
- gamla.string_utils.capitalize(text: str)[source]
Capitalize only the first letter of a given string, unlike str.capitalize that transforms all characters after the first character to lowercase.
>>>capitalize(“my name is John Smith”) ‘My name is John Smith’
- gamla.string_utils.regex_match(pattern)[source]
A curried regex match. Gets a pattern and returns a function that expects a text to match the pattern with.
>>> regex_match(r"phone:(\d*)")("phone:1234567").group(1) # noqa: W605 '1234567'
- gamla.string_utils.replace_in_text(old: str, new: str)[source]
Return a copy of the string with all occurrences of substring old replaced by new >>> txt = “hello world” >>> replace_in_text(“world”, “Jhon”)(txt) ‘hello Jhon’
- gamla.string_utils.split_text(sep: str)[source]
Return a list of the words in the string, using sep as the delimiter string
>>> txt = "hello world" >>> split_text(" ")(txt) ['hello', 'world']
- gamla.tree.filter_leaves(predicate: Callable)[source]
Gets a predicate, and builds a function that gets a dictionary, potentially nested and returns an iterable of leaf values. The values returned are of leafs that pass the predicate.
>>> gamla.pipe({"x": {"y": (1, 2, 3)}}, gamla.filter_leaves(gamla.greater_than(2)), tuple) (3,)
Useful for retrieving values from large json objects, where the exact path is unimportant.
- gamla.tree.get_leaves_by_ancestor_predicate(predicate: Callable)[source]
Gets a predicate, and builds a function that gets a dictionary, potentially nested and returns an iterable of leaf values. The values returned are of leafs where some ancestor (possibly indirect) passes the predicate.
>>> gamla.pipe({"x": {"y": (1, 2, 3)}}, gamla.get_leaves_by_ancestor_predicate(gamla.equals("x")), tuple) (1, 2, 3) >>> gamla.pipe({"x": {"y": (1, 2, 3)}}, gamla.get_leaves_by_ancestor_predicate(gamla.equals("z")), tuple) ()
Useful for retrieving values from large json objects, where the exact path is unimportant.
- gamla.tree.json_tree_reduce(reduce_fn: Callable, tree_node)
Reduce a JSON like tree.
- gamla.tree.map_reduce_tree(children: Callable, reducer: Callable, mapper: Callable)[source]
Like tree_reduce, but allows for an async map stage first, so it can be parallelized.
- gamla.tree.tree_reduce(get_children: Callable, reduce_fn: Callable, tree_node)[source]
Reduces a tree from the bottom up.
Given get_children, a function from a node to its children, and reduce_fn, which gets a node and the results of the reduce on the children, reduces a tree upwards.
- gamla.tree.tree_reduce_async(get_children: Callable, reduce_fn: Callable)[source]
Async version of tree_reduce.
- gamla.type_safety.composable(destination: Callable, origin: Callable, key: str | None) bool[source]
Checks if destination can be composed after source, considering their typing.
- gamla.type_safety.is_subtype(*args, **kwargs)
Given two typings, checks if the second is a superset of the first.