API¶

function_pipe¶

class FunctionNode(function: Any, *, doc_function: Optional[Callable] = None, doc_args: Tuple[Any, ...] = (), doc_kwargs: Optional[Dict[str, Any]] = None)¶

A wrapper for a callable that can reside in an expression of numerous FunctionNodes, or be modified with unary or binary operators.

__init__(function: Any, *, doc_function: Optional[Callable] = None, doc_args: Tuple[Any, ...] = (), doc_kwargs: Optional[Dict[str, Any]] = None) → None¶

Args:

function: a callable or value. If given a value, will create a function that simply returns that value.

doc_function: the function to display in the repr; will be set to function if not provided

doc_args: the positional arguments to display in the repr

doc_kwargs: the keyword arguments to display in the repr

property unwrap: Callable¶: The doc_function should be set to the core function being wrapped, no matter the level of wrapping.

__call__(*args: Any, **kwargs: Any) → Any¶: Call the wrapped function with args and kwargs.

partial(*args: Any, **kwargs: Any) → function_pipe.core.function_pipe.FunctionNode¶: Return a new FunctionNode with a partialed function with args and kwargs.

__neg__() → function_pipe.core.function_pipe.FN¶: Return a new FunctionNode that when evaulated, will negate the result of self

__invert__() → function_pipe.core.function_pipe.FN¶

Return a new FunctionNode that when evaulated, will invert the result of self

NOTE:: This is generally expected to be a Boolean inversion, such as ~ (not) applied to a Numpy, Pandas, or Static-Frame objects.

__abs__() → function_pipe.core.function_pipe.FN¶: Return a new FunctionNode that when evaulated, will find the absolute value of the result of self

__eq__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self==value.

__lt__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self<value.

__le__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self<=value.

__gt__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self>value.

__ge__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self>=value.

__ne__(rhs: Any) → function_pipe.core.function_pipe.FN¶: Return self!=value.

__rshift__(rhs: Callable) → function_pipe.core.function_pipe.FN¶: Composes a new FunctionNode will call lhs first, and then feed its result into rhs

__rrshift__(lhs: Callable) → function_pipe.core.function_pipe.FN¶: Composes a new FunctionNode will call lhs first, and then feed its result into rhs

__lshift__(rhs: Callable) → function_pipe.core.function_pipe.FN¶: Composes a new FunctionNode will call rhs first, and then feed its result into lhs

__rlshift__(lhs: Callable) → function_pipe.core.function_pipe.FN¶: Composes a new FunctionNode will call rhs first, and then feed its result into lhs

__or__(rhs: function_pipe.core.function_pipe.FN) → function_pipe.core.function_pipe.FN¶: Only implemented for PipeNode.

__ror__(lhs: function_pipe.core.function_pipe.FN) → function_pipe.core.function_pipe.FN¶: Only implemented for PipeNode.

class PipeNode(function: Any, *, doc_function: Optional[Callable] = None, doc_args: Tuple[Any, ...] = (), doc_kwargs: Optional[Dict[str, Any]] = None, call_state: Optional[function_pipe.core.function_pipe.PipeNode.State] = None, predecessor: Optional[function_pipe.core.function_pipe.PN] = None)¶

This encapsulates the node that will be used in a pipeline.

It is not expected to be created directly, rather, through usage of pipe_node (and related) decorators.

PipeNodes will be in (or move between) one of three states, depending on where it was created, or what the current state of pipeline evaluation is

partial(*args: str, **kwargs: str) → function_pipe.core.function_pipe.PN¶: Partialing PipeNodes is prohibited. Use pipe_node_factory (and related) decorators to pass in expression-level arguments.

property call_state: Optional[State]¶: The current call state of the Node

property predecessor: Optional[function_pipe.core.function_pipe.PN]¶: The PipeNode preceeding this Node in a pipeline. Can be None

__or__(rhs: function_pipe.core.function_pipe.PN) → function_pipe.core.function_pipe.PN¶: Invokes rhs, passing in self as the kwarg PREDECESSOR_PN.

__ror__(lhs: function_pipe.core.function_pipe.PN) → function_pipe.core.function_pipe.PN¶: Invokes lhs, passing in self as the kwarg PREDECESSOR_PN.

__getitem__(pn_input: Any) → Any¶

Invokes self, passing in pn_input as the kwarg PN_INPUT.

NOTE:

If None, will evaluate self with a default PipeNodeInput instance
If user desires for the initial input to be literally None, use (**{PN_INPUT: None}) instead.

__call__(*args: Any, **kwargs: Any) → Any¶: Call the wrapped function with args and kwargs.

class PipeNodeInput¶

PipeNode input to support store and recall; subclassable to expose other attributes and parameters.

store(key: str, value: Any) → None¶: Store key and value in the underlying store.

recall(key: str) → Any¶: Recall key from the underlying store. Can raise an KeyError

property store_items: ItemsView[str, Any]¶: Return an items view of the underlying store.

pipe_node(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>, self_keyword: str = 'self') → Union[Callable, function_pipe.core.function_pipe.PipeNode]¶

Decorates a function to become a PipeNode that takes no expression-level args.

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

Examples:

>>> @pipe_node
>>> def func(**kwargs):
>>>     pass

>>> @pipe_node()
>>> def func():
>>>     pass

>>> @pipe_node(PN_INPUT)
>>> def func(pn_input):
>>>     pass

>>> class Example:
>>>     @pipe_node(PN_INPUT)
>>>     def method(self, pn_input):
>>>         pass

>>> from functools import partial
>>> español_pipe_node = partial(pipe_node, self_keyword="uno_mismo")
>>> ...
>>> class Ejemplo:
>>>     @español_pipe_node(PN_INPUT)
>>>     def método(uno_mismo, pn_input):
>>>         pass

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.
self_keyword: which keyword to look for when decorating instance methods.

pipe_node_factory(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>, self_keyword: str = 'self') → Union[Callable, Callable[[Any], function_pipe.core.function_pipe.PipeNode]]¶

Decorates a function to become a pipe node factory, that when given expression-level arguments, will return a PipeNode

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

Examples:

>>> @pipe_node_factory
>>> def func(a, b, **kwargs):
>>>     pass
>>> ...
>>> func(1, 2) # This is now a PipeNode!

>>> @pipe_node_factory()
>>> def func(*, a, b):
>>>     pass
>>> ...
>>> func(a=1, b=2) # This is now a PipeNode!

>>> @pipe_node_factory(PN_INPUT, PREDECESSOR_RETURN)
>>> def func(pn_input, previous_value, a, *, b):
>>>     # pn_input will be given the PN_INPUT from the pipeline
>>>     # prev will be given the PREDECESSOR_RETURN from the pipeline
>>>     pass
>>> ...
>>> func(1, b=2) # This is now a PipeNode!

>>> class Example:
>>>     @pipe_node_factory(PN_INPUT, PREDECESSOR_RETURN)
>>>     def method(self, pn_input, previous_value, a, *, b):
>>>         pass
>>> ...
>>> Example().method(1, b=2) # This is now a PipeNode!

>>> from functools import partial
>>> español_pipe_node_factory = partial(pipe_node_factory, self_keyword="uno_mismo")
>>> ...
>>> class Ejemplo:
>>>     @español_pipe_node_factory(PN_INPUT, PREDECESSOR_RETURN)
>>>     def método(uno_mismo, pn_input, valor_anterior, a, *, b):
>>>         pass
>>> ...
>>> Ejemplo().método(1, b=2) # Esto ahora es un PipeNode!

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.
self_keyword: which keyword to look for when decorating instance methods.

compose(*funcs: Callable) → function_pipe.core.function_pipe.FN¶: Given a list of functions, execute them from right to left, passing the returned value of the right f to the left f. Store the reduced function in a FunctionNode

classmethod_pipe_node(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>) → Union[Callable, function_pipe.core.function_pipe.PipeNode]¶

Decorates a function to become a classmethod PipeNode that takes no expression-level args.

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

This is a convenience method, that is the mental equivalent to this pseudo-code:

>>> @classmethod
>>> @pipe_node(...)
>>> def func(...)

Examples:

>>> @classmethod_pipe_node
>>> def func(cls, **kwargs):
>>>     pass

>>> @classmethod_pipe_node()
>>> def func(cls):
>>>     pass

>>> @classmethod_pipe_node(PN_INPUT)
>>> def func(cls, pn_input):
>>>     pass

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.

classmethod_pipe_node_factory(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>) → Callable¶

Decorates a function to become a classmethod pipe node factory, that when given expression-level arguments, will return a PipeNode

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

This is a convenience method, that is the mental equivalent to this pseudo-code:

>>> @classmethod
>>> @pipe_node_factory(...)
>>> def func(...)

Examples:

>>> @classmethod_pipe_node_factory
>>> def func(cls, a, b, **kwargs):
>>>     pass
>>> ...
>>> SomeClass.func(1, 2) # This is now a PipeNode!

>>> @classmethod_pipe_node_factory()
>>> def func(cls, *, a, b):
>>>     pass
>>> ...
>>> SomeClass.func(a=1, b=2) # This is now a PipeNode!

>>> @classmethod_pipe_node_factory(PN_INPUT, PREDECESSOR_RETURN)
>>> def func(cls, pn_input, previous_value, a, *, b):
>>>     # ``pn_input`` will be given the PN_INPUT from the pipeline
>>>     # ``previous_value`` will be given the PREDECESSOR_RETURN from the pipeline
>>>     pass
>>> ...
>>> SomeClass.func(1, b=2) # This is now a PipeNode!

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.

staticmethod_pipe_node(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>) → Union[Callable, function_pipe.core.function_pipe.PipeNode]¶

Decorates a function to become a staticmethod PipeNode that takes no expression-level args.

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

This is a convenience method, that is the mental equivalent to this pseudo-code:

>>> @staticmethod
>>> @pipe_node(...)
>>> def func(...)

Examples:

>>> @staticmethod_pipe_node
>>> def func(**kwargs):
>>>     pass

>>> @staticmethod_pipe_node()
>>> def func():
>>>     pass

>>> @staticmethod_pipe_node(PN_INPUT)
>>> def func(pn_input):
>>>     pass

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.

staticmethod_pipe_node_factory(*key_positions: typing.Union[typing.Callable, str], core_decorator: typing.Callable[[typing.Any], typing.Callable] = <function _core_logger>) → Callable¶

Decorates a function to become a staticmethod pipe node factory, that when given expression-level arguments, will return a PipeNode

This can either be used as a decorator, or a decorator factory, similar to functools.lru_cache.

This is a convenience method, that is the mental equivalent to this pseudo-code:

>>> @staticmethod
>>> @pipe_node_factory(...)
>>> def func(...)

Examples:

>>> @staticmethod_pipe_node_factory
>>> def func(a, b, **kwargs):
>>>     pass
>>> ...
>>> SomeClass.func(1, 2) # This is now a PipeNode!

>>> @staticmethod_pipe_node_factory()
>>> def func(*, a, b):
>>>     pass
>>> ...
>>> SomeClass.func(a=1, b=2) # This is now a PipeNode!

>>> @staticmethod_pipe_node_factory(PN_INPUT, PREDECESSOR_RETURN)
>>> def func(pn_input, previous_value, a, *, b):
>>>     # ``pn_input`` will be given the PN_INPUT from the pipeline
>>>     # ``previous_value`` will be given the PREDECESSOR_RETURN from the pipeline
>>>     pass
>>> ...
>>> SomeClass.func(1, b=2) # This is now a PipeNode!

Args:

key_positions: either a single callable, or a list of keywords that will be positionally bound to the decorated function.
core_decorator: a decorator that will be applied to the core_callable. This is typically a logger. By default, it will print to stdout.

store(pni: PipeNodeInput, ret_val: tp.Any, label: str) → tp.Any:¶: Store ret_val (the value returned from the previous PipeNode) to pni under label. Forward ret_val.

recall(pni: PipeNodeInput, label: str) → tp.Any:¶: Recall label from pni`` and return it. Can raise an KeyError

call(*pns: PipeNode) → tp.Any¶

Broadcasts pns, and returns the result of pns[-1]

Since pns are all PipeNodes, they will all be evaluated before passed in as values to this function.

pretty_repr(f: Any) → str¶: Provide a pretty string representation of a FN, PN, or anything. If the object is a FN or PN, it will recursively represent any nested FNs/PNs.

is_unbound_self_method(core_callable: Union[classmethod, staticmethod, Callable], *, self_keyword: str) → bool¶: Inspects a given callable to determine if it’s both unbound, and the first argument in its signature is self_keyword