nonion 0.3.5

Python Functional Programming for Humans.

NOnion

NOnion is a Python package that provides tools for Functional Programming. One of its aims is to eliminate nested function calls such as z(g(f(x))) which remind an onion.

Installing

pip install nonion

Tutorial

NOnion consists of two submodules:

• nonion.tools - contains a set of functions and types that might simplify your workflow with Functional Programming in Python,
• nonion.loader - contains a wrapper which takes a function Callable[[io.IOBase], X] (such as json.load), and returns a function Callable[[typing.Optional[str]], nonion.Option[X]].

Also NOnion provides two handful tools:

• Function - a wrapper of any Python Callable,
• Pipeline - a wrapper of any Python Iterable.

It is important to understand that NOnion provides tools used for FP in context of Python. Because it is impossible to fully implement some core concepts of FP in Python, NOnion provides tools that resemble other FP languages tools, but are not exactly the same tools.

nonion.tools

Option

Option is a type alias. Option resembles Haskell's Maybe in Python. Option is defined as follows:

Option = Union[Tuple[X], Tuple[()]]

As we can see Option is simply some tuple that might contain a single value or be an empty tuple. It means that in order to initialize an Option you can simply write:

x = () # empty Option
y = (3,) # Option with value 3

You can easily check whether an Option is empty:

def f(x: int) -> Option[int]:
  return (x,) if x < 3 else ()

x: Option[int] = f(5)

if not x:
  print("Option is empty") # Option is empty

You can also provide an alternative value if Option is empty and immediately try to unwrap the Option:

x: Option[int] = f(5)
y, *_ = x or (42,)

print(y) # 42

# alternatively

x: Option[int] = f(1)
z = x or (42,)

# notice: if you pass an empty *z to a single argument function, you will get an error
print(*z) # 1

If you need to apply some function to a content of the Option, you can use nonion.fmap:

x: Option[int] = f(5)
z: Option[int] = fmap(lambda y: y + 1, x)

for i in z:
  print(i)

Because Option is simply a tuple under the hood, you can apply any Python function (that operates on tuple) to an instance of an Option.

Either

Either is a type alias. Either is defined as follows:

Either = Tuple[Option[X], Option[Y]]

Either can be used when you need to return either first value or a second value:

def readline(path: str) -> Either[str, str]:
  buffer: Option[IOBase] = wraptry(open)(path)

  if not buffer:
    return ((), ("error occurred during open",))

  line: Option[str] = fmap(lambda x: x.readline(), buffer)
  fmap(lambda x: x.close(), buffer)

  return (line, ())

line, error = readline("requirements.txt")

if line:
  print(*line)
else:
  print(*error)

Because Either is simply a type alias, it does not checks whether only single value is passed.

as_catch

as_catch is simply:

@curry
def as_catch(default: Callable[[X], Y], xys: Iterable[Tuple[X, Y]]) -> Callable[[X], Y]:
  return catch(as_match(xys), default=default)

Example of as_catch usage:

successor: Callable[[int], int] = Pipeline(range(10)) // zipmapr(lambda x: x + 1) >> as_catch(lambda _: -1)
print(successor(1)) # 2
print(successor(100)) # -1

as_either

as_either is a function which allows you to create an Either[Y, X] from an Option[X]. as_either takes a function Callable[[], Y] and returns a function Callable[[Option[X]], Either[Y, X]].

raw_numbers = "1\n22\nten\n333".splitlines()

xs = (
  Pipeline(raw_numbers)
  / wraptry(int)
  / as_either(lambda: f"Failed to parse.")
  & print
)

# ((), (1,))
# ((), (22,))
# (('Failed to parse.',), ())
# ((), (333,))

The difference between using as_either and explicitly creating Either using tuples is that as_either will not evaluate the left part if the right part is present. That is why Callable[[], Y] is being passed to as_either instead of Y.

as_match

as_match is simply:

def as_match(xys: Iterable[Tuple[X, Y]]) -> Callable[[X], Option[Y]]:
  x_to_y = dict(xys)

  def lookup(x: X) -> Option[Y]:
    return (x_to_y[x],) if x in x_to_y else ()

  return lookup

Example of as_match usage:

successor: Callable[[int], Option[int]] = Pipeline(range(10)) // zipmapr(lambda x: x + 1) >> as_match
print(successor(1)) # (2,)
print(successor(100)) # ()

between

between is simply:

def between(low: float, high: float) -> Callable[[float], bool]:
  return lambda x: low <= x and x <= high

Example of between usage:

ys = filter(between(3, 5), range(10))
print(tuple(ys)) # (3, 4, 5)

binary_compose

binary_compose is an implementation of a ``Function composition" defined as $( f \circ g )(x) = f(g(x))$.

xs = "a", "ab", "c"
yxs = enumerate(xs)

p: Callable[[Tuple[int, str]], bool] = binary_compose(lambda x: x.startswith("a"), second)
filtered: Iterable[Tuple[int, str]] = filter(p, yxs)

ys = map(first, filtered)
print(tuple(ys)) # (0, 1)

bind

bind resembles Haskell's >>= in Python.

def f(x: int) -> Option[int]:
  return (x + 1,) if x < 3 else ()

x: Option[int] = f(1)
y: Option[int] = bind(f, x)

print(*y) # 3

cache

cache is a decorator which returns a function that always returns a value that was returned in the first call.

def f(x: int) -> int:
  return x + 5

g = cache(f)
print(g(5)) # 10
print(g()) # 10
print(g("abc", 1, {})) # 10

h = cache(f)
print(h(7)) # 12

call

call is simply:

def call(fx: Tuple[Callable[..., Y], Tuple[object, ...]]) -> Y:
  f, *x = fx
  return f(*x)

We assume, that Tuple[object, ...] are positional function arguments.

Example of call usage:

def get_initials(name: str, surname: str) -> str:
  return name[:1] + surname[:1]

names = "Haskell Curry", "John Smith", "George Sand"

(
  Pipeline(names)
  // zipl(repeat(get_initials))
  / key(star)
  / value(lambda x: x.split(" "))
  / call
  & print
)

# HC
# JS
# GS

catch

catch is a function that resembles pattern-matching in Python. It takes some functions *fs: Callable[[X], Option[Y]] with some catch-all function default: Callable[[X], Y] and returns a function Callable[[X], Y] which executes fs functions one by one until some function will return non-empty Option[Y]. If none of those functions will return a non-empty Option[Y], the result of default function is returned.

# let's say that we want to parse age ranges that we have in our data:
age_ranges = (
  "10-20",
  "20-30",
  "30+",
  "60+",
  "invalid input"
)

# we consider 30+ to be a valid range <30, 100)

def parse_range(x: str) -> Tuple[int, int]:
  raw = x.split("-")
  low, high, *_ = map(int, raw)

  return low, high

def parse_unbounded_range(x: str) -> Tuple[int, int]:
  raw, *_ = x.split("+")
  return int(raw), 100

# we will use <18, 100) as our default range
parse = catch(
  wraptry(parse_range),
  wraptry(parse_unbounded_range),
  default=lambda _: (18, 100)
)

for x in age_ranges:
  print(parse(x))

# (10, 20)
# (20, 30)
# (30, 100)
# (60, 100)
# (18, 100)

curry

curry is simply:

def curry(f: Callable[..., Y]) -> Callable[..., Y]:
  return lambda *args, **kwargs: partial(f, *args, **kwargs)

drop

drop is simply:

def drop(n: int) -> Callable[[Iterable[X]], Iterable[X]]:
  return lambda xs: islice(xs, n, None)

Example of drop usage:

xs = drop(1)(range(3))
print(tuple(xs)) # (1, 2)

xs = islice(range(3), 1, None)
print(tuple(xs)) # (1, 2)

fail

fail is a function which takes a function Callable[[Exception], Y] and returns a decorator which takes a function Callable[..., Y] and returns Callable[..., Y]. The function returned by the decorator uses passed Callable[[Exception], Y] to handle possible errors produced by a decorated function. If no errors produced, Callable[[Exception], Y] will not be executed and the result of the decorated function will be returned.

# Let's say that you want to write is_repeated function
# which tells you whether you have a collection consisting
# only from the single value.

# The simplest function you could think of might look like this:

def is_repeated(xs: Iterable[X]) -> bool:
  x, *rest = xs
  return all(x == y for y in rest)

# It works on collections that have at least one value:

print(is_repeated((1, 1))) # True
print(is_repeated((1, 2, 3))) # False

# but when you have an empty collection, this function will result
# in an error:

print(is_repeated(()))
# ValueError: not enough values to unpack (expected at least 1, got 0)

# In order to handle this case, you can rewrite this function in a
# following manner:

def is_repeated(xs: Iterable[X]) -> bool:
  xs = iter(xs)
  wrapped_x = wrapnext(xs)

  if wrapped_x:
    x, *_ = wrapped_x
    return all(x == y for y in xs)
  else: return True

# And it would work:

print(is_repeated((1, 1))) # True
print(is_repeated((1, 2, 3))) # False
print(is_repeated(())) # True

# You might also use a *fail* function which will surround your
# function with try-except clause, to deal with empty collection.

@fail(lambda _: True)
def is_repeated(xs: Iterable[X]) -> bool:
  x, *rest = xs
  return all(x == y for y in rest)

# In case when error is raised by is_repeated, the
# lambda _: True
# function will be executed. The raised error will be passed to
# that function.

def g(e: Exception) -> bool:
  print(e)
  return True

@fail(g)
def is_repeated(xs: Iterable[X]) -> bool:
  x, *rest = xs
  return all(x == y for y in rest)

print(is_repeated((1,))) # True
print(is_repeated(()))
# not enough values to unpack (expected at least 1, got 0)
# True

find

find is a function which takes a predicate and returns a function which takes some Iterable and returns an Option with value that matches the predicate if such value exists:

x: Option[int] = find(lambda x: x == 3)(range(5))
print(x) # (3,)

x: Option[int] = find(lambda x: x == -1)(range(5))
print(x) # ()

find_and_collect

find_and_collect is a function which takes a predicate, some Iterator and a buffer, and returns an Option and passed buffer. The Option contains a value that matches the predicate if such value exists. The buffer contains values that were checked using the predicate:

buffer = []
xs = iter(range(5))
x, filled_buffer = find_and_collect(lambda x: x == 3, xs, buffer)

print(x) # (3,)
print(filled_buffer) # [0, 1, 2, 3]

# notice: Iterator has to be passed, not Iterable

buffer = []
x, filled_buffer = find_and_collect(lambda x: x == 3, range(5), buffer)

print(x) # ()
print(filled_buffer) # []

findindex

findindex is a function that works like find, but instead of returning a function which returns a value in Iterable that matches some predicate, it returns a function which returns an index of that value in Iterable.

x: Option[int] = findindex(lambda x: x == 8)(range(5, 10))
print(x) # (3,)

x: Option[int] = findindex(lambda x: x == -1)(range(5, 10))
print(x) # ()

first

first is simply:

def first(xy: Tuple[X, Y]) -> X:
  return xy[0]

flattenl

flattenl is a function which takes a Tuple which contains another Tuple on the beginning and flattens that inner Tuple inside of outer Tuple.

xys = {"A": 2.5, "B": 3.14}
Pipeline(xys.items()) // zipr(count(1)) / flattenl & print

# ('A', 2.5, 1)
# ('B', 3.14, 2)

flattenr

flattenr is a function which takes a Tuple which contains another Tuple on the end and flattens that inner Tuple inside of outer Tuple.

xys = {"A": 2.5, "B": 3.14}
Pipeline(xys.items()) // zipl(count(1)) / flattenr & print

# (1, 'A', 2.5)
# (2, 'B', 3.14)

fleft

fleft is a function which allows you to flatten an Either which contains another Either as its right value. It is simply defined as:

@curry
def fleft(f: Callable[[Z], X], xzy: Either[X, Either[Z, Y]]) -> Either[X, Y]:
  x, zy = xzy

  if x:
    return x, ()
  else:
    (z, y), *_ = zy
    return fmap(f, z), y

Example of fleft usage:

def drive(x: int) -> Either[str, str]:
  if x < 18: return ("You are too young to drive.",), ()
  else: return (), (("OK",),)

f: Callable[[str], Either[str, Tuple[()]]] = lambda x: (
  Pipeline(wrapexcept(int)(x))
  // left(lambda y: f"Parsing error: {y}")
  // right(drive)
  // fleft(lambda y: f"Drive error: {y}")
  >> tuple
)

print(f("eighteen")) # (("Parsing error: invalid literal for int() with base 10: 'eighteen'",), ())

print(f("17")) # (('Drive error: You are too young to drive.',), ())

print(f("25")) # ((), (('OK',),))

flip

flip is simply:

def flip(f: Callable[[Y, X], Z]) -> Callable[[X, Y], Z]:
  return lambda x, y: f(y, x)

Example of flip usage:

xs = "A", "B", "C"
Pipeline(enumerate(xs)) / key(lambda x: x + 1) * star(flip(repeat)) & print

# A
# B
# B
# C
# C
# C

fmap

fmap resembles Haskell's fmap in Python. It is intended to be used with Option, because it transforms the result of Python's map function into tuple. fmap is defined as follows:

def fmap(f: Callable[[X], Y], x: Iterable[X]) -> Tuple[Y, ...]:
  return binary_compose(tuple, lift(f))(x)

If you simply want to lift some function without composing the resulting function with a tuple, use a lift function.

def f(x: int) -> Option[int]:
  return (x + 1,) if x < 3 else ()

x: Option[int] = f(1)
y: Option[int] = fmap(lambda x: x + 5, x)

print(*y) # 7

foldl

foldl is a function which takes a binary function Callable[[Y, X], Y] and some accumulator Y and returns a function which takes Iterable[X] and returns Y. This function allows you to fold Iterable[X] from left using passed binary function. The accumulator is being passed as the first argument of the binary function.

Example of foldl usage:

xs = range(ord("A"), ord("Z") + 1)
alphabet = Pipeline(xs) / chr >> foldl(operator.add, "")

print(alphabet)

# ABCDEFGHIJKLMNOPQRSTUVWXYZ

foldl1

foldl1 is a similar function to foldl. The difference between foldl1 and foldl is that foldl1 takes Callable[[X, X], X], uses Iterable[X] first element as the accumulator and returns X. foldl1 will raise an error if the supplied Iterable[X] is empty.

foldr

foldr is a function which takes a binary function Callable[[X, Y], Y] and some accumulator Y and returns a function which takes Iterable[X] and returns Y. This function allows you to fold Iterable[X] from right using passed binary function. The accumulator is being passed as the last argument of the binary function.

Example of foldr usage:

xs = range(ord("A"), ord("Z") + 1)
reversed_alphabet = (
  Pipeline(xs)
  / chr
  // foldr(lambda x, acc: acc + [x], [])
  >> foldl(operator.add, "")
)

print(reversed_alphabet)

# ZYXWVUTSRQPONMLKJIHGFEDCBA

foldr1

foldr1 is a similar function to foldr. The difference between foldr1 and foldr is that foldr1 takes Callable[[X, X], X], uses Iterable[X] last element as the accumulator and returns X. foldr1 will raise an error if the supplied Iterable[X] is empty. Under the hood foldr1 will use tuple on passed Iterable[X] in order to extract the accumulator.

group

group is a function which takes Iterable[X] and returns Iterable[Tuple[X, ...]]. This function groups passed elements by equality comparison ==.

xs = 1, 1, 2, 2, 2, 3, 1, 1, 1
print(tuple(group(xs))) # ((1, 1), (2, 2, 2), (3,), (1, 1, 1))

groupby

groupby is a function which takes an equality comparison function Callable[[X, X], bool] and returns a function Callable[[Iterable[X]], Iterable[Tuple[X, ...]]] which groups passed elements by the equality comparison function.

people = (
  ("Alex", 23),
  ("John", 23),
  ("Sam", 27),
  ("Kate", 27),
  ("Fred", 23),
)

grouped = groupby(lambda x, y: second(x) == second(y))(people)
print(tuple(grouped))
# ((('Alex', 23), ('John', 23)), (('Sam', 27), ('Kate', 27)), (('Fred', 23),))

# or you can use *on* function:

grouped = groupby(on(operator.eq, second))(people)
print(tuple(grouped))
# ((('Alex', 23), ('John', 23)), (('Sam', 27), ('Kate', 27)), (('Fred', 23),))

in_

in_ is simply:

def in_(xs: Tuple[X, ...]) -> Callable[[X], bool]:
  return lambda x: x in xs

iterfind

iterfind is a function which takes an Iterable of predicates and returns a function which takes some Iterable and returns an Iterable of Option. Each Option contains a matched value of a corresponding predicate. iterfind uses find_and_collect under the hood. iterfind firstly searches for matching value in a buffer, if it could not find one, it passes predicate along with buffer to find_and_collect.

fs = (lambda x: x == 2), (lambda x: x == 4), (lambda x: x == 1), (lambda x: x == -1)
ys: Iterable[Option[int]] = iterfind(fs)(range(5))

for y in ys:
  print(y)

# (2,)
# (4,)
# (1,)
# ()

key

key is simply:

def key(f: Callable[[X], Z]) -> Callable[[Tuple[X, Y]], Tuple[Z, Y]]:
  g: Callable[[Tuple[X, Y]], Z] = binary_compose(f, first)
  return lambda xy: (g(xy), second(xy))

Example of key usage:

xys = {"A": [1, 2, 3], "B": [3, 4]}
zys = map(key(str.casefold), xys.items())

for zy in zys:
  print(zy)

# ('a', [1, 2, 3])
# ('b', [3, 4])

left

left is simply:

@curry
def left(f: Callable[[X], Z], xy: Either[X, Y]) -> Either[Z, Y]:
  x, y = xy
  return fmap(f, x), y

length

length is a function which takes an Iterable and returns number of elements in that Iterable. length exhausts the Iterable.

xs = 1, 2, 3
print(len(xs)) # 3

# len(iter(xs)) will raise an error
print(length(iter(xs))) # 3

lift

lift is simply:

lift = curry(map)

maptry

maptry is a function which takes a function Callable[[X], Y], and returns some function which takes Iterable[X] and returns Iterable[Y]. If a function Callable[[X], Y] will raise an error for some element, that element will be omitted.

possible_jsons = "{}", "", "123, 32323", "{\"a\": 1}"
jsons = maptry(json.loads)(possible_jsons)

for x in jsons:
  print(x)

# {}
# {'a': 1}

match

match is a function that resembles pattern-matching in Python. It takes some functions *fs: Callable[[X], Option[Y]] and returns a function Callable[[X], Option[Y]] which executes fs functions one by one until some function will return non-empty Option[Y]. If none of those functions will return a non-empty Option[Y], an empty Option[Y] (i.e. ()) is returned.

# let's say that we want to parse age ranges that we have in our data:
age_ranges = (
  "10-20",
  "20-30",
  "30+",
  "60+",
  "invalid input"
)

# we consider 30+ to be a valid range <30, 100)

def parse_range(x: str) -> Tuple[int, int]:
  raw = x.split("-")
  low, high, *_ = map(int, raw)

  return low, high

def parse_unbounded_range(x: str) -> Tuple[int, int]:
  raw, *_ = x.split("+")
  return int(raw), 100

parse = match(
  wraptry(parse_range),
  wraptry(parse_unbounded_range)
)

for x in age_ranges:
  print(parse(x))

# ((10, 20),)
# ((20, 30),)
# ((30, 100),)
# ((60, 100),)
# ()

merge

merge is a function which takes two sorted Iterable[X] and merges them into single sorted Iterable[X]. It uses lambda x, y: x <= y comparison function by default. The comparison function might be substituted with any other function of type signature Callable[[X, X], bool] by passing a compare parameter.

xs = merge((1, 3, 5), (1, 2, 4))
print(tuple(xs)) # (1, 1, 2, 3, 4, 5)

not_

not_ is a function which takes a predicate and returns negation of that predicate.

print(not_(lambda x, y: x == y)(1, 5)) # True

on

on is simply:

def on(f: Callable[[Y, Y], Z], g: Callable[[X], Y]) -> Callable[[X, X], Z]:
  return lambda p, n: f(g(p), g(n))

Example of on usage could be found in groupby section.

partition

partition is a function which takes a predicate and returns a function Callable[[Iterable[X]], Tuple[Tuple[X, ...], Tuple[X, ...]]]. This returned function splits passed elements into those that do match the predicate and the rest. The difference between span and partition is that span stops when it finds the first element that does not match the predicate and partition goes until the end.

xs = 1, 1, 2, 2, 2, 3, 1, 1, 1
matched, rest = partition(lambda x: x == 1)(xs)

print(matched) # (1, 1, 1, 1, 1)
print(rest) # (2, 2, 2, 3)

pick

pick is a function which takes some aggregate function Callable[[Tuple[X, ...]], X] and returns a function Callable[[Iterable[X]], Iterable[X]]. This returned function picks all elements from the passed collection Iterable[X] which are equal to the value returned by the aggregate function. The equal function might be substituted with any other function of type signature Callable[[X, X], bool] by passing a compare parameter.

print(min([1, 2, 1, 1, 3, 1])) # 1

ys = tuple(pick(min)([1, 2, 1, 1, 3, 1]))
print(ys) # (1, 1, 1, 1)

print(min([]))
# ValueError: min() arg is an empty sequence

ys = tuple(pick(min)([]))
print(ys) # ()

pickby

pickby is a function which takes a function Callable[[X], Y], an aggregate function Callable[[Tuple[Y, ...]], Y] and returns a function Callable[[Iterable[X]], Iterable[X]]. This returned function picks all elements from the passed collection Iterable[X] which corresponding Y values, created by the Callable[[X], Y] function, are equal to the value returned by the aggregate function. The equal function might be substituted with any other function of type signature Callable[[Y, Y], bool] by passing a compare parameter. The function Callable[[X], Y] will be used exactly once on the whole collection.

cars_and_prices = (
  ("Audi", 25000),
  ("BMW", 70000),
  ("Mercedes", 25000),
)

cheapest_car = min(cars_and_prices, key=second)
print(cheapest_car) # ('Audi', 25000)

cheapest_cars = pickby(second, min)(cars_and_prices)
print(tuple(cheapest_cars)) # (('Audi', 25000), ('Mercedes', 25000))

powerset

powerset is a function which takes a Tuple[X, ...] and produces power set of those elements in form of Iterable[Iterable[X]].

xs = tuple(range(3))
ps = tuple(map(tuple, powerset(xs)))

print(ps) # ((), (2,), (1,), (1, 2), (0,), (0, 2), (0, 1), (0, 1, 2))

replicate

replicate is a function which takes a number n and returns a function, which takes some value x and repeats n times value x.

xs = tuple(replicate(5)("hello"))
print(xs)
# ('hello', 'hello', 'hello', 'hello', 'hello')

reverse

reverse is a function which takes an Iterable[X] and returns Deque[X] which contains elements from Iterable[X] in reversed order.

xs = range(ord("A"), ord("Z") + 1)

reversed_alphabet = (
  Pipeline(xs)
  / chr
  // reverse # Python reversed would not work on Iterable
  >> foldl(operator.add, "")
)

print(reversed_alphabet)

# ZYXWVUTSRQPONMLKJIHGFEDCBA

right

right is simply:

@curry
def right(f: Callable[[Y], Z], xy: Either[X, Y]) -> Either[X, Z]:
  x, y = xy
  return x, fmap(f, y)

scanl

scanl is a similar function to foldl. The difference between scanl and foldl is that scanl instead of returning a function which takes Iterable[X] and returns Y, returns a function which takes Iterable[X] and returns Iterable[Y]. The resulting Iterable[Y] contains all accumulators used in foldl.

xs = scanl(operator.mul, 1)((1, 2, 3, 4, 5))
print(tuple(xs))
# (1, 1, 2, 6, 24, 120)

scanl1

scanl1 is a similar function to foldl1. The difference between scanl1 and foldl1 is that scanl1 instead of returning a function which takes Iterable[X] and returns X, returns a function which takes Iterable[X] and returns Iterable[X]. The resulting Iterable[X] contains all accumulators used in foldl1.

xs = scanl1(operator.mul)((1, 2, 3, 4, 5))
print(tuple(xs))
# (1, 2, 6, 24, 120)

scanr

scanr is a similar function to foldr. The difference between scanr and foldr is that scanr instead of returning a function which takes Iterable[X] and returns Y, returns a function which takes Iterable[X] and returns Deque[Y]. The resulting Deque[Y] contains all accumulators used in foldr.

xs = scanr(operator.mul, 1)((1, 2, 3, 4, 5))
print(xs)
# deque([120, 120, 60, 20, 5, 1])

scanr1

scanr1 is a similar function to foldr1. The difference between scanr1 and foldr1 is that scanr1 instead of returning a function which takes Iterable[X] and returns X, returns a function which takes Iterable[X] and returns Deque[X]. The resulting Deque[X] contains all accumulators used in foldr1.

xs = scanr1(operator.mul)((1, 2, 3, 4, 5))
print(xs)
# deque([120, 120, 60, 20, 5])

search

search is a function which takes a predicate Callable[[X], bool] along with Iterable[Y] and returns a function which takes Iterable[X] and returns Iterable[Y]. This function zips Iterable[Y] with Iterable[X] and returns those Ys for which corresponding Xs match the predicate.

xs = search(lambda x: x > 3, count())(range(1, 6))
print(tuple(xs))
# (3, 4)

second

second is simply:

def second(xy: Tuple[X, Y]) -> Y:
  return xy[1]

shift

shift is a decorator which returns a partially applied function. The difference between Python's functools.partial and shift is that shift will return a function which prepends *args and **kwargs:

def dummy(*args: object, **kwargs: object):
  print(args)
  print(kwargs)

partial(dummy, 1, 2, a=1, b="b")(3, 4, c="c")
print("-" * 10)
shift(dummy, 1, 2, a=1, b="b")(3, 4, c="c")

# (1, 2, 3, 4)
# {'a': 1, 'b': 'b', 'c': 'c'}
# ----------
# (3, 4, 1, 2)
# {'c': 'c', 'a': 1, 'b': 'b'}

Example of shift usage:

take_3 = shift(islice, 3)
xs: Iterable[int] = take_3(range(5))

for x in xs:
  print(x)

# 0
# 1
# 2

slide

slide is a function which takes a sliding window length n and a step, and returns a function which takes an Iterable and applies sliding window over it resulting in an Iterable of tuples. Each tuple has at most length equal to n. In case when strict=True option is passed, each tuple has length equal to n. step is simply a shift of a sliding window.

xs: Iterable[Tuple[int, ...]] = slide()(range(10))
print(tuple(xs))
# ((0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, 6), (6, 7), (7, 8), (8, 9), (9,))

xs: Iterable[Tuple[int, ...]] = slide(n=3, step=2)(range(10))
print(tuple(xs))
# ((0, 1, 2), (2, 3, 4), (4, 5, 6), (6, 7, 8), (8, 9))

xs: Iterable[Tuple[int, ...]] = slide(n=3, step=2, strict=True)(range(10))
print(tuple(xs))
# ((0, 1, 2), (2, 3, 4), (4, 5, 6), (6, 7, 8))

def is_sorted(xs: Iterable[X], compare: Callable[[X, X], bool] = operator.le) -> bool:
  return (
    Pipeline(slide(strict=True)(xs))
    / star(compare)
    >> all
  )

print(is_sorted((1, 2, 5))) # True
print(is_sorted((1, 2, -5))) # False

span

span is a function which takes a predicate and returns a function Callable[[Iterable[X]], Tuple[Tuple[X, ...], Iterable[X]]]. This returned function splits passed elements into those that do match the predicate on the beginning and the rest.

xs = 1, 1, 2, 2, 2, 3, 1, 1, 1
matched, rest = span(lambda x: x == 1)(xs)

print(matched) # (1, 1)
print(tuple(rest)) # (2, 2, 2, 3, 1, 1, 1)

splitat

splitat is a function which takes an index i and returns a function which splits an Iterable[X] into Tuple[X, ...] and Iterable[X]. The Tuple[X, ...] will contain first i elements and the Iterable[X] will contain the rest.

xs, rest = splitat(1)(range(5))
print(xs) # (0,)
print(tuple(rest)) # (1, 2, 3, 4)

strip

strip is a function which takes an Iterable[X] and returns an Iterable[X] with removed consecutive duplicates. strip functions uses only equality comparison ==.

xs = 1, 1, 2, 2, 2, 3, 1, 1, 1
print(tuple(strip(xs))) # (1, 2, 3, 1)

stripby

stripby is a function which takes an equality comparison function Callable[[X, X], bool] and returns a function Callable[[Iterable[X]], Iterable[X]] which removes consecutive duplicates in terms of the equality comparison function.

people = (
  ("Alex", 23),
  ("John", 23),
  ("Sam", 27),
  ("Kate", 27),
  ("Fred", 23),
)

stripped = stripby(lambda x, y: second(x) == second(y))(people)
print(tuple(stripped))
# (('Alex', 23), ('Sam', 27), ('Fred', 23))

# or you can use *on* function:

stripped = stripby(on(operator.eq, second))(people)
print(tuple(stripped))
# (('Alex', 23), ('Sam', 27), ('Fred', 23))

take

take is simply:

def take(n: int) -> Callable[[Iterable[X]], Iterable[X]]:
  return lambda xs: islice(xs, n)

Example of take usage:

xs = take(1)(range(3))
print(tuple(xs)) # (0,)

xs = islice(range(3), 1)
print(tuple(xs)) # (0,)

tee

tee is a decorator which allows you to apply some function to a passed argument and return back the passed argument instead of a function's result.

x = tee(print)(5) # 5
print(x) # 5

xs = (
  Pipeline(range(3, 0, -1))
  / tee(print, "Countdown:", file=sys.stderr)
  >> tuple
)
# Countdown: 3
# Countdown: 2
# Countdown: 1

print(xs) # (3, 2, 1)

value

value is simply:

def value(f: Callable[[Y], Z]) -> Callable[[Tuple[X, Y]], Tuple[X, Z]]:
  g: Callable[[Tuple[X, Y]], Z] = binary_compose(f, second)
  return lambda xy: (first(xy), g(xy))

Example of value usage:

xys = {"A": [1, 2, 3], "B": [3, 4]}
xzs = map(value(len), xys.items())

for xz in xzs:
  print(xz)

# ('A', 3)
# ('B', 2)

where

where is a similar function to findindex. The difference between where and findindex is that where returns indices of all elements that match given predicate instead of one. The other difference is that where returns a function which takes Iterable[X] and returns Iterable[Y], on the other hand findindex returns a function which takes Iterable[X] and returns Option[int].

xs = where(lambda x: x >= 8)(range(5, 10))
print(tuple(xs)) # (3, 4)

xs = where(lambda x: x == -1)(range(5, 10))
print(tuple(xs)) # ()

wrapeek

wrapeek is a function which takes an Iterable and returns an Option containing a first value of the Iterable along with an original Iterable (containing first value).

xs = (x for x in range(5))
x, ys = wrapeek(xs)

print(x) # (0,)

for y in ys:
  print(y)

# 0
# 1
# 2
# 3
# 4

wrapexcept

wrapexcept is a decorator which returns a function that returns Either with some value or an Exception that was raised.

f = wrapexcept(next)
xs = iter(range(2))

print(f(xs)) # ((), (0,))
print(f(xs)) # ((), (1,))
print(f(xs)) # ((StopIteration(),), ())

wrapnext

wrapnext is simply:

wrapnext: Callable[[Iterator[X]], Option[X]] = wraptry(next)

Example of wrapnext usage:

xs = iter(range(2))

print(wrapnext(xs)) # (0,)
print(wrapnext(xs)) # (1,)
print(wrapnext(xs)) # ()

wraptry

wraptry is a decorator which returns a function that returns Option with some value or an empty Option if an Exception was raised.

load_json = wraptry(json.loads)

print(load_json("{}")) # ({},)
print(load_json("[1, 2, 3]")) # ([1, 2, 3],)
print(load_json("abc")) # ()

zipflatl

zipflatl is a function which takes a function Callable[[X], Option[Y]], and returns some function which takes Iterable[X] and returns Iterable[Tuple[X, Y]] with only those elements from Iterable[X] that are mapped to non-empty Option[Y] by the function Callable[[X], Option[Y]].

xs = "1", "hello", "2"
f = wraptry(int)

ys = zipflatl(f)(xs)
print(tuple(ys)) # ((1, '1'), (2, '2'))

zipflatr

zipflatr is a function which takes a function Callable[[X], Option[Y]], and returns some function which takes Iterable[X] and returns Iterable[Tuple[Y, X]] with only those elements from Iterable[X] that are mapped to non-empty Option[Y] by the function Callable[[X], Option[Y]].

xs = "1", "hello", "2"
f = wraptry(int)

ys = zipflatr(f)(xs)
print(tuple(ys)) # (('1', 1), ('2', 2))

zipl

zipl is simply:

def zipl(xs: Iterable[X]) -> Callable[[Iterable[Y]], Iterable[Tuple[X, Y]]]:
  return lambda ys: zip(xs, ys)

Example of zipl usage:

xs = "A", "B", "C"
Pipeline(xs) // zipl(count(1)) * star(flip(repeat)) & print

# A
# B
# B
# C
# C
# C

zipmapl

zipmapl is simply:

def zipmapl(f: Callable[[X], Y]) -> Callable[[Iterable[X]], Iterable[Tuple[Y, X]]]:
  return lambda xs: map(lambda x: (f(x), x), xs)

Example of zipmapl usage:

xs = range(ord("a"), ord("z") + 1)
upper_to_lower = Pipeline(xs) / chr // zipmapl(str.upper) >> dict

Pipeline(upper_to_lower.items()) // take(5) & print

# ('A', 'a')
# ('B', 'b')
# ('C', 'c')
# ('D', 'd')
# ('E', 'e')

zipmapr

zipmapr is simply:

def zipmapr(f: Callable[[X], Y]) -> Callable[[Iterable[X]], Iterable[Tuple[X, Y]]]:
  return lambda xs: map(lambda x: (x, f(x)), xs)

Example of zipmapr usage:

xs = range(ord("a"), ord("z") + 1)
upper_to_lower = Pipeline(xs) / chr // zipmapr(str.upper) >> dict

Pipeline(upper_to_lower.items()) // take(5) & print

# ('a', 'A')
# ('b', 'B')
# ('c', 'C')
# ('d', 'D')
# ('e', 'E')

zipr

zipr is simply:

def zipr(ys: Iterable[Y]) -> Callable[[Iterable[X]], Iterable[Tuple[X, Y]]]:
  return lambda xs: zip(xs, ys)

Example of zipl usage:

xys = {"A": 2.5, "B": 3.14}
Pipeline(xys.items()) // zipr(count(1)) / flattenl & print

# ('A', 2.5, 1)
# ('B', 3.14, 2)

nonion.loader

FROM_STDIN

FROM_STDIN is None. FROM_STDIN is defined for readability purposes. When you write CLI which can read users input from stdin by default, you can use this constant instead of using None.

as_loader

as_loader is a decorator which takes a BufferLoader and creates a Loader. BufferLoader and Loader are defined as follows:

BufferLoader = Callable[[IOBase, Tuple[object, ...], Dict[str, object]], X]
Loader = Callable[[Optional[str], Tuple[object, ...], Dict[str, object]], Option[X]]

For example, json.load and pd.read_csv are BufferLoader's.

Created Loader will take a path as its first argument and will read the content using Python built-in open. If path is not provided, Loader reads content from stdin. If during read or during BufferLoader call exception raises, Loader will return an empty Option.

# first_column_extractor.py
from typing import Callable, Optional

import pandas as pd

from nonion import Option
from nonion import as_loader
from nonion import bind
from nonion import fmap
from nonion import wraptry

load_frame = as_loader(pd.read_csv)
frame: Option[pd.DataFrame] = load_frame()

get_first_column = wraptry(lambda x: x.iloc[:, 0])
# x.iloc[:, 0] might raise an error, so use wraptry

series: Option[pd.Series] = bind(get_first_column, frame)

to_csv = lambda x: x.to_csv(header=False, index=False)
raw_series: Option[str] = fmap(to_csv, series)

if not series:
  print("something went wrong")
else:
  print(*raw_series, end="")

We can use script first_column_extractor.py in a following way in a Bash-like shells:

python first_column_extractor.py < frame.csv

load

load is a function which takes an Optional path to a file and returns an IOBase buffer containing content of the file. If path does not exists load uses stdin.

with load() as buffer:
  print(buffer.read())

Notice: when load uses stdin, it firstly reads whole stdin content.

load_json

load_json is simply:

load_json: Loader[Union[Dict[str, object], Tuple[object, ...]]] = as_loader(json.load)

Example of load_json usage:

x = load_json("object.json")
print(x) # ([1, 2, 3],)

wrapopen

wrapopen is simply:

wrapopen: Callable[[str], Option[IOBase]] = wraptry(open)

Example of wrapopen usage:

x = wrapopen("missing_object.json")
print(x) # ()

Function

In order to create a Function, you simply pass any Callable:

f = Function(lambda x: x + 1)
f(5) # returns 6

You can also create an identity Function:

g = Function()

Notice, that a Function takes exactly single value and returns exactly single value.

compose

A ``Function composition" defined as $( f \circ g )(x) = f(g(x))$ could be done in the following way:

z = f @ g

# alternatively

z = f.compose(g)

You can also use compose several times:

z = f @ g @ f

Instead of wrapping each Callable with a Function, you can wrap only first Callable and use compose on the rest.

def f(x):
  return x + 1

g = Function() @ (lambda x: x * 2) @ f
g(5) # returns 12

The @ (at) operator was used, because it reminds $\circ$ symbol.

then

Function composition sometimes might be hard to read, because you have to read it from right-to-left. In order to achieve better readability, you can use then.

g = Function() / (lambda x: x * 2) / f
g(5) # returns 11

# alternatively

g = Function().then(lambda x: x * 2).then(f)
g(5) # returns 11

The / (slash) operator was used, because it reminds | (vertical bar) used for piping.

call

Sometimes you want to call a function ``inline'' after several compositions. In this case, you might use:

(Function() / (lambda x: x * 2) / f)(5) # returns 11

But it might be hard to read. Especially, when you mostly pass lambdas. A better way to call a function is by using:

Function() / (lambda x: x * 2) / f & 5 # returns 11

The & (ampersand) operator was used, because it looks similar to $ (dollar), which is a Haskell operator.

star (function)

Suppose, that you defined a function with multiple arguments such as:

def f(x, y):
  return x + y * x

And you want to wrap that function using Function. In this case, you have to use star.

Function() @ star(f) & (1, 2) # returns 5

star simply passes arguments to a function using Python * (star) operator.

unstar (function)

unstar is the opposite function to star:

names = unstar(", ".join)("Haskell Curry", "John Smith", "George Sand")
print(names) # Haskell Curry, John Smith, George Sand

foreach

You can also call a function for each value in some Iterable in the following way:

ys = Function() / (lambda x: x * 2) / (lambda x: x + 1) * range(5)

for y in ys:
  print(y)

# 1
# 3
# 5
# 7
# 9
#

The * (star) operator was used, because instead of passing an Iterable to a function, you pass its content as with Python * (star) operator and functions that take *args.

Pipeline

In order to create a Pipeline, you simply pass any Iterable:

xs = Pipeline(range(5))

# notation abuse, do not use that:

xs = Function() / Pipeline & range(5)

You can also create an empty Pipeline:

xs = Pipeline()

Under the hood Pipeline is simply uses iter on a passed Iterable. It means, that if you will pass an Iterable, that could be exhausted, you iterate over Pipeline only once.

xs = Pipeline(range(2))

for x in xs:
  print(x)

# 1
# 2
#

# perfectly fine, because range(x) returns a special object
for x in xs:
  print(x)

# 1
# 2
#

xs = Pipeline(x for x in range(2))

for x in xs:
  print(x)

# 1
# 2
#

# xs already exhausted
for x in xs:
  print(x)

map

map allows you to call a Callable, which takes a single value and returns a single value, on each value of the Pipeline.

ys = Pipeline(range(3)) / (lambda x: x + 1) / (lambda x: (x, x + 1)) / star(lambda x, y: x + y * x)

for y in ys:
  print(y)

# 3
# 8
# 15
#

# alternatively

ys = Pipeline(range(3)).map(lambda x: x + 1).map(lambda x: (x, x + 1)).map(star(lambda x, y: x + y * x))

The / (slash) operator was used, because it reminds | (vertical bar) used for piping.

filter

filter allows you to filter Pipeline values.

ys = Pipeline(range(3)) % (lambda x: x > 1)

for y in ys:
  print(y)

# 2
#

# alternatively

ys = Pipeline(range(3)).filter(lambda x: x > 1)

flatmap

flatmap allows you to call a Callable, which takes a single value and returns an Iterable, on each value of the Pipeline and flatten results into single Pipeline.

ys = Pipeline(range(2)) / (lambda x: x + 1) * (lambda x: (x, x + 1))

for y in ys:
  print(y)

# 1
# 2
# 2
# 3
#

# alternatively

ys = Pipeline(range(2)).map(lambda x: x + 1).flatmap(lambda x: (x, x + 1))

The * (star) operator was used, because intuitively you use Python * (star) operator on each result.

apply

apply allows you to call a Callable, which takes an Iterable and returns an Iterable, on whole Pipeline.

ys = Pipeline(range(2)) / (lambda x: x + 1) // tuple # internally Pipeline now has a tuple

for y in ys:
  print(y)

# 1
# 2
#

# now multiple itertations is possible
for y in ys:
  print(y)

# 1
# 2
#

# alternatively

ys = Pipeline(range(2)).map(lambda x: x + 1).apply(tuple)

collect

collect allows you to call a Callable, which takes an Iterable and returns any single value, on whole Pipeline. The difference between apply and collect is that collect returns the result of a function instead of wrapping it with Pipeline.

ys = Pipeline(range(2)) / (lambda x: x + 1) >> tuple
print(ys)

# (1, 2)
#

# alternatively

ys = Pipeline(range(2)).map(lambda x: x + 1).collect(tuple)

You can also combine collect with any function which takes an Iterator:

ys = Pipeline(range(2)) / (lambda x: x + 1) >> wrapnext
print(ys) # (1,)

ys = Pipeline(range(2)) % (lambda x: x == 5) >> wrapnext
print(ys) # ()

ys = Pipeline(range(5)) >> shift(islice, 2)

for y in ys:
  print(y)

# 0
# 1

# alternatively you can use apply

ys = Pipeline(range(5)) // shift(islice, 2) & print

# 0
# 1

foreach

foreach allows you to call a Callable, which takes a single value, on each value of the Pipeline.

Pipeline(range(2)) / (lambda x: x + 1) & print

# 1
# 2
#

# alternatively

Pipeline(range(2)).map(lambda x: x + 1).foreach(print)

groupon

groupon is a function which takes a function Callable[[X], Y], and returns some function which takes Iterable[X] and returns Iterable[X] grouped on Callable[[X], Y] function. The groupon function uses Python groupby function under the hood. groupon adds a grouping key using passed Callable[[X], Y] function and sorts values by that key before applying groupby.

xs = -3, 1, 0, -1, 5

(
  Pipeline(xs)
  // groupon(lambda x: x > 0)
  / value(tuple)
  & print
)

# (False, (-3, 0, -1))
# (True, (1, 5))