Applicatives, or functors for arbitrary functions¶
Mapping only single param functions can be limiting. It would be cool if lift would could work on function with arbitrary many parameters.
But before we begin with our attempts at solution, we need to define our fmap and lift once again:
from functools import singledispatch, partial
@singledispatch
def fmap(x, fn):
raise Error("Not implemented for" + a)
def lift(fn):
def liftedfn(x):
return fmap(x, fn)
return liftedfn
Now we could register a few classes that would work with this. I.e. all of the collections such as list:
@fmap.register([].__class__)
def _(lst, fn):
return [fn(x) for x in lst]
Could currying rescue us?
Currying is a concept, where instead of passing in all the params into the function at once, you pass in the first param and grt a new function as a result. This function accepts the second param and returns a function that accepts the third, and by now you should be getting a picture.
Or maybe I should show an example.Lets invent a nontrivial 3 arg function, we will call it erm, window (because slice was taken).
from functools import partial
def window(begin,end,array):
return array[begin:end]
window(2,4,[1,2,3,4,5])
If this function were curried, it would look like this:
def curried_window(begin):
return lambda end: (lambda array: array[begin:end])
curried_window(2)(4)([1,2,3,4,5])
Some functional languages have this as a default. The nice thing about that is, that you can quickly create more specific fuctions, without the need to wrap them in new function definitions.
And this seems to solve out problem with multi param function, because sudenly, everything is a single param function, once you curry :)
fmap([1,2,3,4],curried_window)
Ok, as you can see, currying won't save us just yet, because even though we can use map with them, result is kind-of useless list of functions. We would need to somehow be able to apply this list of functions on a list of values.
@singledispatch
def apply(lx, lfn):
raise Error("Not implemented for" + lx)
@fmap.register([].__class__)
def apply(lx,lfn):
return [fn(x) for fn in lfn for x in lx]
a0 = map(curried_window,[1,2])
a1 = apply([5,6],a0)
a2 = apply([
[1,2,3,4,5,6],
[1,2,3,4,5,6]
],a1)
a2
Yay, this seems to have worked :)
Now we can try to combine currying with apply and create a propper lift function :)
def curry(n, fn):
if n == 1:
return fn
if n == 2:
return lambda x:partial(fn,x)
else:
return lambda x:curry(n-1,partial(fn,x))
def lift(fn):
def lifted(arg0, *args):
result_fn = map(curry(len(args)+1, fn),arg0)
for a in args:
result_fn = apply(a, result_fn)
return result_fn
return lifted
There is one more assumption we made, when creating this lift function. On the line where we state:
result = map(curry(len(args)+1, fn),arg0)we assume that you can use map function on the data-type we are lifting.
liftedw = lift(window)
liftedw([1,2], [5,6], [
[1,2,3,4,5,6],
[1,2,3,4,5,6],
])
So, why to through all of this trouble to get the equivalent of i.e:
[window(x,y,z)
for x in [1,2]
for y in [5,6]
for z in [[1,2,3,4,5,6],[1,2,3,4,5,6]]]
I have two reasons:
- We could use this pattern for things that are not iterable :)
- we can analyze the pattern in isolation and figure out interesting relations
Lifting values¶
So far, we were concerned about lifting functions, but we would probably like to be able to lift arbitrary values as well. As you can see in our lifted definition, we have so far assumed, that we are lifting at least an unary function. Because we can't derive the wrapper for values from map and apply, we need to define it for the specific datatype. We uaually call this function pure
def pure(value):
return [value]
Interesting thing with pure is, that with it defined, we should be able to replace uses of map(fn,o) with apply(pure(fn),o).
Algebra behind the applicative¶
So far we have not been explicit about our formal requirements on map, applicative and pure. I have mentioned that map should behave sensible and that there is some relationship between functors, lifting and category theory.
In my mind, the most sensible thing, is to think about functor as a lawful interface. Lawful interface would mean, that it conforms to more requirements, than just the ability to i.e. perform a method call.
A type T describes a functor if:
- we can consider T a wrapper over arbitrary type A
- there is a function
map, that- accepts function
f, that acceptsaand returnsb - accepts
Twrappinga - returns
Twrappingb
- accepts function
In haskell we would say that map has type (a -> b) -> T a -> T b
With Python's typing library we could say that the type of map is
Callable[[Callable[[A],B], T[A]],T[B]]The map function then needs to conform to these laws
- for any t,
map(lambda x:x ,t) == t - for any t and any two composable functions f and g,
map(f,map(g,t)) == map(lambda x: f(g(x)),t)
If we define id and compose functions:
def id(x):
return x
def compose(f,g):
def composed(x)
return f(g(x))
return composedwe can swap the lambda expressions in the laws:
- for any t,
map(id ,t) == t - for any t and any two composable functions f and g,
map(f,map(g,t)) == map(compose(f,g),t)
For applicative, we need it to implement be functor as well as
- have function
purethat accepts arbitraryaand returntT a - have function
apply, that- for any function f, that accepts
aand returnsb - accepts
Twrappingf - accepts
Twrappinga - returns
Twrappingb
- for any function f, that accepts
In haskell, we would say, that pure has type a -> T a and apply T (a -> b) -> T a -> T b
With Python's typing library we could say that the type of pure is
Callable[[A], T[A]]And type of apply is
Callable[[T[Callable[[A],B]], T[A]],T[B]]The apply and pure functions need to conform to these laws:
- for any t,
apply(pure(id),t) == t - for any function f and type x,
apply(pure(f),pure(x)) == f(x) - for any t, x,
apply(t, pure(x)) == apply(pure(lambda g: g(x)),t) - for any u,v,w,
apply(apply(apply(pure(compose),u),v),w) == apply(u,apply(v,w))- be aware that apply is not required to be associative, i.e. you can't state what ?? should be in
apply(apply(u,v),w) ?? apply(u,apply(v,w))
- be aware that apply is not required to be associative, i.e. you can't state what ?? should be in
Parametrized lift¶
For some of the following examples, I need to be able to lift into different domains. I might want to do this by making the lift function parametrized. We can't unfortunately do this with single dispatch, because the pure function would have nothing to dispatch on. This is similiar to our problem with empty when dealing with monoids.
This is why we will make applicative a class
class Applicative:
def pure(self, val):
raise NotImplementedError();
def apply(self, fn, val):
raise NotImplementedError();
def lift(self, fn):
def lifted(arg0, *args):
result = self.apply(self.pure(curry(len(args)+1, fn)),arg0)
for a in args:
result = self.apply(result, a)
return result
return lifted
Because applicatives are not native to python, I am not entirely sure how to best define them. But this way looks simple enough.
Solving the null-pointer exception?¶
One thing we could try, is to solve the NPE problem with this. If you think about it, when you have function, that doesn't expect None, but gets it, only reasonable thing we could do with this, is return None.
We can define our wrapper function pure to just return the value back. We wouldn't be able to do this kind of cheating in haskell, where we would need to explicitly box the type. In Python this seems to be working.
We define our apply function to return None if any of the params are None.
We can try this on int function.
class ApplyNone(Applicative):
def pure(self, val):
return val
def apply(self, lfn,lx):
if lfn!=None and lx!=None:
return lfn(lx)
return None
nint = ApplyNone().lift(int)
Now we can compare:
print(int("1"))
print(int(None))
print(nint("1"))
print(nint(None))
To be honest, this is sort of a toy example. You probably don't want to just wrap your functions with this and instead of error at the place of failure just recieve the propagated nulls everywhere. But we can do better.
Applicative Validation¶
So what about some other uses? One I really like is using applicatives for validation.
First, let me harp on a thing I don't like about exceptions. Once you throw it, you are aborting the execution. I especially dislike it, if somebody uses exceptions to validate function params, i.e.:
def validated_window(begin, end, arr):
if begin < 0:
raise Exception("Begin was negative")
if end < 1:
raise Exception("End was not positive")
return window(begin,end,arr)
validated_window(-1,-1,[])
Exception only told me about the frst problem. But with applicative pattern, I can rewrite it in such way, that I would get all of them.
I will define two validating functions:
def valid_begin(begin):
return begin if begin >= 0 else Exception("Begin was negative")
def valid_end(end):
return end if end > 0 else Exception("End was not positive")
Now, let me define how would I like to write the validated function.
def validated_window(begin, end, arr):
return liftedw(valid_begin(begin),valid_end(end),arr)
Now the interesting part of the puzzle is to redefine applyy once again, from thing that processes values in arrays, to thing that can deal with values that might be useless Exceptions.
class ApplyError(Applicative):
def pure(self, val):
return val
def apply(self,lfn,lx):
if isinstance(lfn,Exception) and isinstance(lx,Exception):
return Exception(*lfn.args,*lx.args)
if isinstance(lfn,Exception):
return lfn
if isinstance(lx,Exception):
return lx
return lfn(lx)
As you can see, there are 3 cases to consider:
- if both fn and x contain exceptions, I return agregate exception
- if fn or x contain exceptions, I return the exception
- if none of these happen, I can return fn(x)
liftedw = ApplyError().lift(window);
validated_window(-1,-1,[])
This style of applicative validation is more or less taken from the Purescript book. There is a difference, in the explicit boxing and unboxing I have mentioned earlier. If we explixitly boxed i.e. the Exception in an object of i.e. Validated class, we might be able to use some nice features of python, like sinlge dispatch. On the other hand, the lack of static type checking means, that the boxing wouldn't really help us.
Applicative generators¶
In python we have the concept of generator, that just produces a stream values. For example, we could have a infinite stream of positive and negative numbers.
def positive():
i = 0;
while True:
i+=1
yield i
def negative():
i=0
while True:
i1=1
yield i
Previously we created applicative, that would be able to lift iterables/generators by doing the equivalent of:
def apply(wf,wx)
return [f(x)
for f in wf
for x in wx]This would of course only work for generators that end. But we could combne generators in a way that would be equal to:
def apply(wf,wx):
return (f(x)
for f,x in zip(wf, wx))
So, how would we define the pure function? One way might be:
def pure(v):
yield v
There is a problem with this definition, though. We want apply(pure(id),t) == t, for any t. And if we try i.e. t = [1, 2, 3]
list(apply(pure(lambda x:x),[1,2,3]))
This means that the correct definition for pure will need to be an infinite stream, repeating the value.
class ApplyGenerator(Applicative):
def pure(self, val):
while True:
yield val
def apply(self, wf, wx):
return (f(x)
for f,x in zip(wf, wx))
Now we can try this with our window function.
list(ApplyGenerator().lift(window)([1,2], [5,6], [
[1,2,3,4,5,6],
[1,2,3,4,5,6],
]))
And it seems that it is working as expected :-)
Composing applicatives¶
Now, after we have defined several somewhat useful applicatives, we can nest them. I am stealing the implementation from the original paper Applicative programming with effects from the part Composing applicative functors
def containing(outer, inner):
class Nested(Applicative):
def pure(self, val):
return outer.pure(inner.pure(val))
def apply(self, lfn, lval):
return outer.lift(inner.apply)(lfn,lval)
return Nested()
Now we can make a i.e. a stream of possible errors.
ListOfValid = containing(ApplyGenerator(), ApplyError() )
You can see, that this still works with the vaules it have worked previously.
list(ListOfValid.lift(window)([1,2], [5,6], [
[1,2,3,4,5,6],
[1,2,3,4,5,6],
]))
But we can try to pass in some error values alongside of the iterables.
list(ListOfValid.lift(window)([1,Exception("Begin was negative")], [5,Exception("End was not positive")], [
[1,2,3,4,5,6],
[1,2,3,4,5,6],
]))
You can see that the first values of the iterables were correctly combined into result, and second values that had some errors have the combined error result.
list(ListOfValid.lift(window)([Exception("Begin was negative"),Exception("Begin was negative")], [5, 6], [
[1,2,3,4,5,6],
[1,2,3,4,5,6],
]))
Here we have error in both the first and the second and it results in both errors being shown in the resulting stream.