nonion 0.2.3

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

AnyFunction

AnyFunction is a type alias that describes any Python function. AnyFunction has following two assumptions:

• Tuple[object, ...] - is interpreted as args,
• Dict[str, object] - is interpreted as kwargs.

AnyFunction is defined as follows:

AnyFunction = Callable[[Tuple[object, ...], Dict[str, object]], object]

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_function(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_function

as_function is simply:

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

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

  return lookup

Example of as_function usage:

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

between

between is simply:

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

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

cachepartial

cachepartial is simply:

def cachepartial(f: AnyFunction[Y], *args: object, **kwargs: object) -> AnyFunction[Y]:
  f = partial(f, *args, **kwargs)
  return cache(f)

Example of cachepartial usage:

def f(x: int, y: int) -> int:
  return x + y

g = cachepartial(f, 5)
print(g(5)) # 10

call

call is simply:

def call(fx: Tuple[Callable[[Tuple[object, ...]], 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(explode(get_initials, names))
  / 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: AnyFunction[Y]) -> AnyFunction[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)

explode

explode is simply:

def explode(x: X, ys: Iterable[Y]) -> Iterable[Tuple[X, Y]]:
  return zip(repeat(x), ys)

Example of explode usage:

def multiply(scalar: int, vector: Iterable[int]) -> Iterable[int]:
  xs_and_ys: Iterable[Tuple[int, int]] = explode(scalar, vector)
  return starmap(operator.mul, xs_and_ys)

xs: Iterable[int] = multiply(2, (3, 4, 5))
print(tuple(xs)) # (6, 8, 10)

find

find is a function which takes a predicate and 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 simply:

def findindex(p: Callable[[X], bool], xs: Iterable[X]) -> Option[int]:
  yxs = enumerate(xs)

  g: Callable[[Tuple[int, X]], bool] = binary_compose(p, second)
  yx: Option[Tuple[int, X]] = find(g, yxs)

  return fmap(first, yx)

Example of findindex usage:

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]

flatten

flatten is simply:

def flatten(xyz: Tuple[Tuple[X, Y], Z]) -> Tuple[X, Y, Z]:
  (x, y), z = xyz
  return x, y, z

Example of flatten usage:

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

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

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

fold

fold is simply:

def fold(f: Callable[[Y, X], Y], acc: Y) -> Callable[[Iterable[X]], Y]:
  return lambda xs: reduce(f, xs, acc)

It is a convenience function which takes only swapped second and third arguments of Python's reduce function. The first argument of reduce function has to be supplied to returned function. It makes it easy to partially apply some function and accumulator.

Example of fold usage:

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

print(alphabet)

# ABCDEFGHIJKLMNOPQRSTUVWXYZ

fold1

fold1 is a strict version of fold. It is simply defined as:

def fold1(f: Callable[[Y, X], Y]) -> Callable[[Iterable[X]], Y]:
  return lambda xs: reduce(f, xs)

fold1 resulting function will raise an error if supplied Iterable[X] is empty.

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 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])

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)

mapexplode

mapexplode is simply:

def mapexplode(f: Callable[[X, Y], Z], x: X, ys: Iterable[Y]) -> Iterable[Z]:
  xs_and_ys: Iterable[Tuple[X, Y]] = explode(x, ys)
  return starmap(f, xs_and_ys)

Example of mapexplode usage:

def multiply(scalar: int, vector: Iterable[int]) -> Iterable[int]:
  return mapexplode(operator.mul, scalar, vector)

xs: Iterable[int] = multiply(2, (3, 4, 5))
print(tuple(xs)) # (6, 8, 10)

maptry

maptry is simply:

def maptry(f: Callable[[X], Y], xs: Iterable[X]) -> Iterable[Y]:
  ys: Iterable[Option[Y]] = map(wraptry(f), xs)
  return chain.from_iterable(ys)

Example of maptry usage:

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),)
# ()

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)

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))

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 an Iterable, length and step and returns Iterable of tuple after applying a sliding window. Each tuple has at most length equal to length. 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(range(10), length=3, step=2)
print(tuple(xs))
# ((0, 1, 2), (2, 3, 4), (4, 5, 6), (6, 7, 8), (8, 9))

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)

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,)

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)

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)) / flatten & 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))