DEV Community: Sune Debel

How To Make Functional Programming in Python Go Fast

Sune Debel — Wed, 19 May 2021 19:50:00 +0000

I've previously written about the Python library pfun that allows you to write Python code in a purely functional style by providing a functional effect system. Simply put, the pfun effect system works by representing side-effects as a recursive data-structure that can then be interpreted at the edge of your program to produce real effects, very much inspired by how effects are managed in Haskell or the purely functional ZIO Scala library.

In this post, we'll study how such an effect system can be implemented in Python, how it impacts performance of Python programs, and why the pfun effect system has particularly good performance. We wont spend any time on why you would want to manage side-effects with an effect system. If you want to read more about that, check out one of my previous posts.

To illustrate how the pfun effect system works, let's define our own toy effect system that models interaction with the console. Each interaction with the console will be represented by a type that can be interpreted to produce an effect. To that end, lets define a super-type that all console effects will inherit from:

class Console:
    def interpret(self):
        raise NotImplementedError()

The first effect that comes to mind when thinking about console interaction is printing. Let's define a type that represents console printing:

from dataclasses import dataclass


@dataclass
class PrintLine(Console):
    line: str
    rest: Console

    def interpret(self):
        print(self.line)
        return self.rest.interpret()

Notice that PrintLine has a field rest that allows you to compose it with other console effects. When a PrintLine is interpreted, it prints something to the console and continues interpretation of the composed console effect.

Next, let's handle reading from the console:

from typing import Callable


@dataclass
class ReadLine(Console):
    rest: Callable[[str], Console]

    def interpret(self):
        return self.rest(input()).interpret()

Similarly to PrintLine, ReadLine has a callable field rest that allows you to continue the effect interpretation with the result of reading from the console.

Finally, we need a way to terminate a console program. Lets define a leaf type that just wraps a value, but doesn't continue interpretation recursively:

from typing import Any


@dataclass
class Return(Console):
    val: Any

    def interpret(self):
        return self.val

We can use our model of console interaction to build data-structures that express sequences of operations that interacts with the console:

program = PrintLine(
    'what is your name? ',
    ReadLine(
        lambda name: PrintLine(f'Hello {name}!',
                               Return(None))
    )
)

program on its own is simply a Python value. During its definition, no side-effects are actually carried out. In fact, nothing really happens until program.interpret() is invoked, at which point all the effects represented by program are carried out sequentially.

Using an effect system of this sort to express side-effects obviously comes with the overhead of interpreting the effect data-structure in Python, which may be significant depending on the depth of the effect data-structure.

But what's worse is that interpreting PrintLine and ReadLine instances requires recursively interpreting its sub effects. Since this recursive call is in tail position in both cases, it could in theory be tail-call-optimized away, but alas Python famously doesn't perform tail-call-optimization. This means that a sufficiently deep data-structure will eventually overflow the Python stack.

Concretely, the following program:

p = PrintLine('', Return(None))
for _ in range(10000):
    p = PrintLine('', p)
p.interpret()

Causes the following error:

---------------------------------------------------------------------------
RecursionError                            Traceback (most recent call last)

----> 1 p.interpret()

... in interpret(self)
      5     def interpret(self):
      6         print(self.line)
----> 7         return self.rest.interpret()
      8
      9

... last 1 frames repeated, from the frame below ...

... in interpret(self)
      5     def interpret(self):
      6         print(self.line)
----> 7         return self.rest.interpret()
      8
      9

RecursionError: maximum recursion depth exceeded while calling a Python object

The solution to this problem is to use another data-structure called a trampoline that can interpret the recursive function calls in constant stack space:

@dataclass
class Trampoline:
    def interpret(self):
        raise NotImplementedError()


@dataclass
class Done(Trampoline):
    val: Any

    def interpret(self):
        return self.val


@dataclass
class Call(Trampoline):
    thunk: Callable[[], Trampoline]

    def interpret(self):
        trampoline = self
        while not isinstance(trampoline, Done):
            trampoline = trampoline.thunk()
        return trampoline.val

With this in hand we can re-write our Console effect system as follows:

@dataclass
class Return(Console):
    val: Any

    def interpret(self):
        return Done(self.val)


@dataclass
class PrintLine(Console):
    line: str
    rest: Console

    def interpret(self):
        print(self.line)
        return Call(lambda: self.rest.interpret())


@dataclass
class ReadLine(Console):
    rest: Callable[[str], Console]

    def interpret(self):
        return Call(lambda: self.rest(input()).interpret())

Notice how all the implementations of interpret returns immediately, instead of directly performing a recursive call. This allows us to interpret Console values of arbitrary depth without running into RecursionError.

Unfortunately, this solution makes the performance of our Console effect system even worse, since now we've also added the overhead of interpreting the trampoline data-structure.

The effect system in pfun is conceptually similar to the Console effect system we've built here (although it can do much, much more than interacting with the console). To compose effects in the pfun effect system, we use a function called and_then which takes as an argument a callback function from the result of an effect to a new effect. for example:

from pfun import files


read_foo = files.read('foo.txt')
read_foo.and_then(lambda content: files.write('bar.txt', content))

read_foo.and_then(...) produces a new effect that is similar in concept to PrintLine and ReadLine in that, when interpreted, it will first interpret read_foo, pass the result to the callback function, and then interpret the effect returned by that callback.

Interpreting deep effect data-structures produced by long sequences of calls to and_then would lead to RecursionError just as with our initial version of the Console effect system, were it not for the fact that the pfun effect system also uses a trampolining mechanism in its own interpreter. Unfortunately, running this trampolined interpreter in Python is pretty slow because it involves looping over Python values and calling Python functions, both of which are not terribly fast.

Luckily, Python integrates nicely with C code through a feature called extension types. Running an interpreter for an effect system in C is much, much faster than running it in Python.

In pfun 0.12.0 the interpreter for the effect system was completely re-written as a Python C extension. Lets do some benchmarking to see how big a difference this actually makes in terms of performance. We'll use the performance of the the Scala library ZIO as a baseline, as the pfun effect system draws most of its inspiration from there. ZIO has a fairly extensive benchmarking suite. The most obvious benchmark for testing raw interpreter speed, without any parallelism, is called deepLeftBind (bind is the canonical name for the and_then operation, also called flatMap in Scala):

import zio._

def deepLeftBind(): UIO[Int] = {
    var i  = 0
    var io = IO.effectTotal(i)
    while (i < 10000) {
        io = io.flatMap(i => IO.effectTotal(i))
        i += 1
    }
    io
}

unsafeRun(deepLeftBind())

The equivalent in the pfun effect system looks like this:

from pfun.effect import success


def deep_left_bind():
    i = 0
    e = success(i)
    while i < 10000:
        e = e.and_then(success)
        i += 1
    return e

deep_left_bind().run(None)

Each benchmark is executed 100 times in a loop in a single Python interpreter and Scala REPL respectively, measuring execution time in milliseconds each time. For the ZIO version, 5 warmup executions are also performed before the actual benchmarking loop to prevent time used for JVM class loading etc. to influence to the outcome.

Below I've plotted the mean execution time in milliseconds over the 100 executions for three benchmarks:

pfun 0.11.5 which uses a pure Python interpreter for its effect system
pfun 0.12.1 which uses a C interpreter for its effect system
ZIO 1.0.8

Clearly, pfun 0.12.1 is much, much faster than version 0.11.5. Both ZIO and pfun 0.12.1 are so much faster than pfun 0.11.5 that it's hard to see how pfun 0.12.1 and ZIO compare to each-other. Let's zoom in on just those two:

It's not surprising that there's still some room for improvement in the performance of pfun compared to ZIO, as a lot of work has gone into optimizing the ZIO interpreter. Maybe pfun will never catch up to the performance of ZIO as some of the overhead of the pfun implementation is fundamental to having a Python interface: In order to work with pfun from the Python side, pfun must accept Python callback functions. Jumping back and forth between Python and C during effect interpretation introduces overhead that will be difficult to get rid of, while still keeping the pfun api ergonomical. That being said, the current C implementation of the pfun effect system is only the first version, so there is almost without a doubt room for further optimization of the interpreter itself.

To learn more about pfun, check out the documentation or head over to the github repository.

Completely Type-Safe Dependency Injection in Python

Sune Debel — Thu, 20 Aug 2020 18:28:44 +0000

Simply put, dependency injection is a collection of programming techniques that enables software components to have their dependencies replaced, thereby increasing re-usability, for example by allowing a database module to depend on a connection string in order that it can connect to multiple databases. Dependency injection also improves testability because complicated dependencies such as database connections can be easily replaced with mocks.

In this post we'll study dependency injection in Python. We'll see how it can be made completely type safe using functional programming, specifically with a modern functional programming library called pfun.

Dependency injection in Python is typically implemented using monkey patching. In fact, monkey patching as a form of providing dependencies is so common that the unittest.mock.patch function was added to the standard library in Python 3.3.

While monkey patching is a simple technique, it often leads to rather complex patching scenarios, where it's tricky to figure out what and how to patch. Moreover, it can be tricky to achieve complete type safety with monkey patching on both the dependency consumer and provider side. Finally, there are no straight-forward ways of making sure that the dependency provider provides all required dependencies, often leading to statements such as if dependency is not None.

An alternative (and potentially even simpler) method for dependency injection is the following: function arguments. In fact, this approach is so simple that using the term "dependency injection" to describe it seems almost pretentious. An attractive feature of implementing dependency injection with functions that take arguments is that it is completely type safe by default (provided that you use type annotations of course). Moreover, you can tell a function's dependencies simply by reading its signature, making it totally clear what needs to be provided, and in fact making it impossible to not provide all required dependencies.

The main drawback from using function arguments as your "injection" mechanism, is that functions that call functions that have "injected dependencies" now need to take those dependencies as arguments themselves.

For example, consider a function connect that returns a connection to a database given a connection_str as argument:

def connect(connection_str: str) -> Connection:
    ...

To achieve the promise of "dependency injection", every function that calls connect must now take a connection_str parameter in addition to its other parameters, in order that the calling function itself can be used against different databases:

def get_user(connection_str: str, user_id: int) -> User:
    connection = connect(connection_str)
    return connection.get_user(user_id)

For functions with many dependencies, this quickly becomes very tedious.

Thankfully, we can use functional programming to improve the situation by a margin: Notice how the only use get_user has of connection_str is to pass it to connect. With functional programming we can abstract this pattern of functions taking arguments only to pass them to other functions into some general 'plumbing' functions. This will alow us to write get_user without mentioning the connection_str at all!

To keep things nice and readable (and save us some typing), lets define a type-alias that represents functions that require dependencies:

from typing import TypeVar, Callable


R = TypeVar('R')
A = TypeVar('A')
Depends = Callable[[R], A]

In words, Depends[R, A] is a function that depends on something of type R to produce something of type A. For example, our previous connect function is a Depends[str, Connection] value: it's a function that depends on a str to produce a Connection.

Simple enough, but how do we use the Depends represented by connect in get_user without explicitly taking connection_str as a parameter and calling connect? To do that, let's introduce our first plumbing function map_:

B = TypeVar('B')


def map_(d: Depends[R, A], f: Callable[[A], B]) -> Depends[R, B]:
    def depends(r: R) -> B:
        return f(d(r))
    return depends

map_ precisely implements the "passing of parameters" plumbing we were talking about earlier: it creates a new Depends value that calls d, and passes the result to a function f. This allows us to write get_user as follows:

def get_user(user_id: int) -> Depends[str, User]:
    return map_(connect, lambda con: con.get_user(user_id))

And voila: we have type-safe dependency injection using only Depends (which is just a type-alias for functions of 1 argument) and map_ (which is just a function that composes Depends values with functions). Realise that we could keep applying map_ to Depends values to return a final Depends all the way back to where we make desicions about what concrete connection strings to pass in (probably in the main section of our program). The pattern we've discovered here seems simple enough that it would be surprising if we were the firsts to think of it, and sure enough it's in fact a common pattern in functional programming called the reader effect.

That's all well and good, but what happens when we want to map_ functions that return new Depends values? For example, let's imagine that in addition to reading user data from databases, our application also needs to call an HTTP api with the user data. Doing so requires authentication credentials that we want to inject. The function that calls the api might look like this:

from typing import Tuple


Auth = Tuple[str, str]


def call_api(user: User) -> Depends[Auth, bytes]:
    def depends(auth: Auth) -> bytes:
        ...
    return depends

We might try to use map_ to pass the result of the Depends value returned by get_user to call_api. But in doing so we would end up with a Depends[str, Depends[Auth, bytes]]:

if __name__ == '__main__':
    user_1: Depends[str, User] = get_user(user_id=1)
    call_api_w_user_1: Depends[str, Depends[Auth, bytes]] = map_(user_1, call_api)

When we want to call call_api_w_user_1, we need to first supply the connection string, and then the auth information:

result: bytes = call_api_w_user_1('user@prod')(('prod_user', 'pa$$word'))

This might not seem like big issue in this example, but notice that we might apply map_ over several functions that produces Depends values with the same dependency type, resulting in a situation where we have to pass in the same dependency many times. We might also have a varying number of dependencies if we use map_ in a loop, leading to a situation where we don't know how many times we need to call the final Depends!

To fix this situation, let's introduce another plumbing function that can pass dependencies to both a Depends value, and a Depends value returned by a function that's being mapped. We'll call it and_then since it chains together results of Depends values with functions that return new Depends values:

from typing import Any


R1 = TypeVar('R1')


def and_then(d: Depends[R, A], f: Callable[[A], Depends[R1, B]]) -> Depends[Any, B]:
    def depends(r: Any) -> B:
        return map_(d, f)(r)
    return depends

The observant reader will notice that there's a big problem with the typing of our and_then function: it returns a Depends[Any, B]! Here you might object: 'But wait a minute, I though you said this was "completely type safe"'! The reason for this epic fail is that the r parameter passed to the depends function must be passed to both d, which means it must be of type R, and the Depends returned by f, which means it must be of type R1. In other words, the dependency must be of both type R and R1 at the same time. In our example, using and_then to combine get_user and call_api should result in a type that is both a str and a Tuple[str, str]:

call_api_w_user_1: Depends[?, bytes] = and_then(get_user(user_id=1), call_api)

Such a type is called an intersection type and unfortunately it's not supported by the Python typing module (yet). So without introducing third party libraries, this is the best we can do.

This is where the library pfun comes into the picture. In pfun, map_ and and_then are instance methods of a Depends type, but the idea is exactly the same:

from pfun import Depends


def connect() -> Depends[str, Connection]:
    ...

def get_user(user_id: int) -> Depends[str, User]:
    connect().map(lambda con: con.get_user(user_id))

def call_api(user: User) -> Depends[Auth, bytes]:
    ...

call_api_w_user_1 = get_user(user_id=1).and_then(call_api)

pfun provides mechanisms for combining Depends values with different dependency types when using MyPy with one small requirement: the dependency type must be a typing.Protocol. The reason is that intersections of protocol types are simply a new protocol that inherits from both:

from typing import Protocol


class P1(Protocol):
    pass


class P2(Protocol):
    pass


class Intersection(Protocol, P1, P2):
    pass

This means that when using and_then with Depends values that have protocol types as dependencies, pfun can infer an intersection type automatically. Let's rewrite our example to take advantage of this feature:

from typing import Protocol, Tuple


class HasConnectionStr(Protocol):
    connection_str: str


class HasAuth(Protocol):
    auth: Tuple[str, str]


def connect() -> Depends[HasConnectionStr, Connection]:
    ...


def get_user(user_id: int) -> Depends[HasConnectionStr, Connection]:
    connect().map(lambda con: con.get_user(1))


def call_api(user: User) -> Depends[HasAuth, bytes]:
    ...


call_api_w_user_1 = get_user(user_id=1).and_then(call_api)

The type of call_api_w_user_1 in this example will be Depends[pfun.Intersection[HasConnectionStr, HasAuth], bytes]. The dependency type pfun.Intersection ensures that call_api_w_user_1 must be
called with an argument that fulfills both the HasConnectionStr protocol and the HasAuth protocol. In other words, this would be a type error:

class Dependency:
    def __init__(self):
        connection_str = 'user@prod'


call_api_w_user_1(Dependency())  # MyPy Error!

This compositional nature of and_then means that complex dependency types are built from simple dependency types automatically, which means you can always figure out which dependencies a part of your application has, simply by inspecting it's return type. To learn more about functional programming with pfun check out some of the other posts in the series, the documentation or the github repo.

suned / pfun

Functional, composable, asynchronous, type-safe Python.

Background vector created by macrovector - www.freepik.com

Completely Type-Safe Error Handling in Python

Sune Debel — Tue, 04 Aug 2020 15:20:19 +0000

In this post in my series on functional programming in Python with pfun,
we'll take a look at the Python type system, and how it can be used to make error handling with pfun completely type safe. In a previous post in this series, we introduced the pfun.effect.Effect type and discussed how it models both successful computations (through the A type parameter called the success type) and errors (through the E type parameter called the error type).

Recall that Effect represents functions that:

Take exactly one argument of type R
Returns a result of type A or raises an error of type E
May or may not perform side effects

Let's investigate how this abstraction enables completely type-safe error handling using a small example. In order to understand how Effect enables this feature, let's re-examine the method and_then that chains together an Effect with a function that returns a new Effect:

from typing import TypeVar, Generic, Union, Callable, Any


R = TypeVar('R', contravariant=True)
E = TypeVar('E', covariant=True)
A = TypeVar('A', covariant=True)

E2 = TypeVar('E2')
B = TypeVar('B')


class Effect(Generic[R, E, A]):
    def __call__(self, r: R) -> A
        ...

    def and_then(self, 
                 f: Callable[[A], Effect[Any, E2, B]]
                 ) -> Effect[Any, Union[E, E2], B]:
        ...

We've left out the implementation since it's not particularly important (you can see the details in the previous post if you want). What is important is the signature of and_then. To see how the type of and_then enables type-safe error handling, consider this code:

first: Effect[R, E, A]
second: Effect[R, E2, B]

f = lambda _: second
last: Effect[R, Union[E, E2], B] = first.and_then(f)

result: B  = last(...)

When last is called in the last line, it calls first which might fail with a value of type E. If first succeeds, last calls f with the result of first. f returns second, which is also called called by last. But second might fail with a value of type E2, which ultimately means that last can fail in exactly two ways:

If first fails
if second fails

And so, the correct error type of last must be Union[E, E2], which is what you'll find expressed in the signature of and_then.

The use of Union in the return type of and_then means that complex error types are built up by combining effects with simple error types automatically. This is extremely useful because it allows us to reason about complex errors of combined effects with the help of a type-checker like MyPy without any special effort on our part: we can simply describe effects with simple error types, combine them with and_then (or any other function in the pfun.effect api that accepts multiple effects) and let the type-checker do the complicated work of keeping track of which errors may be raised when an Effect is called.

Automatic tracking of error types isn't particularly useful if we can't eliminate error types by handling them with type safety. pfun.effect.Effect provides several ways of handling errors. We'll study one in particular through the following example:

Imagine you want to use pfun to call an HTTP api and parse the result as JSON in functional style. To make HTTP requests you can use pfun.http:

from http.client import HTTPException

from pfun.effect import Effect
from pfun.http import HTTP

http = HTTP()
call_api: Effect[object, HTTPException, bytes] = http.get('http://foo-api.com')

(Of course you don't have to type out the type of call_api as it can be inferred by MyPy, we do it here only because it's instructive to look at the types).

Since not all byte strings are valid JSON data, let's write a function parse_json that parses JSON data as an Effect. We can do this using the effect.success function that creates an effect that does nothing but return its argument, and effect.error that creates an effect that does nothing but fail with its argument:

from json import JSONDecodeError, loads

from pfun.effect import success, error


def parse_json(s: bytes) -> Effect[object, JSONDecodeError, dict]:
    try:
        return success(loads(b))
    except JSONDecodeError as e:
        return error(e)

Or simply using the effect.catch decorator:

from json import JSONDecodeError, loads

from pfun.effect import catch


parse_json = catch(JSONDecodeError)(loads)

With parse_json in hand, our entire program looks like:

from typing import Union


program: Effect[object, Union[HTTPException, JSONDecodeError], dict]
program = call_api.and_then(parse_json)

Now, imagine that we want to be sure that all errors are handled at the time program is assigned its value. The function we'll use to eliminate errors is called Effect.recover. This function allows us to pass in a function that inspects the error and returns a new effect (which might fail in new ways).

Let's demonstrate by handling the HTTPException. This might be done by calling a backup api. In this example, we'll simply imagine that we can use a default backup response

from typing import NoReturn


def handle_http_error(reason: HTTPException) -> Effect[object, NoReturn, bytes]:
    return success(b'{}')


program: Effect[object, JSONDecodeError, dict] = call_api.recover(handle_http_error).and_then(parse_json)

typing.NoReturn is a special type that indicates that handle_http_error can't return a value for the E type variable of Effect, which is let's the type checker verify that handle_http_error can't fail.

We could of course proceed and handle the JSONDecodeError in a similar fashion, but I think you get the idea.

In summary, pfun.effect enables type safe error handling through pure functional programming without any special effort from the developer. This is useful because it enables the type checker to verify that our error handling is correct (at least in terms of the types). To learn more about pfun check out the documentation or head over to the github repository.

suned / pfun

Functional, composable, asynchronous, type-safe Python.

Be More Lazy, Become More Productive

Sune Debel — Thu, 23 Jul 2020 19:24:21 +0000

In this oxymoronically titled article, we study laziness as a core aspect of functional programming in Python. I'm not talking about hammock driven development or some such leisurely titled paradigm, but lazy evaluation. We'll see how lazy evaluation can make you more productive by improving re-usability and composeability through refactoring a small example, introducing lazyness along the way.

Simply put, lazy evaluation means that expressions are not evaluated until their results are needed. Contrast this with eager evaluation, which is the norm in imperative programming. Under eager evaluation, functions immediately compute their results (and perform their side-effects) when they are called. As an example, consider this python function called get_json which calls a web api as a side-effect and parses the response as json:

import urllib.request
import json


def get_json(url: str) -> dict
    with urllib.request.urlopen(url) as response:
        content = response.read()
        return json.loads(content)

Now, imagine that we want to implement a retry strategy with a simple back-off mechanism. We could take the eager approach and adapt get_json:

import time


def get_json(url: str, retry_attempts: int = 3) -> dict:
    last_exception: Exception
    for attempt in range(1, retry_attempts + 1):
        try:
            with urllib.request.urlopen(url) as response:
                content = response.read()
                return json.loads(content)
        except Exception as e:
            time.sleep(attempt)
            last_exception = e
    raise last_exception

That works, but the solution has a huge shortcoming: We can't re-use the retry strategy for other types of HTTP requests. Or alternatively, but equivalently: get_json violates the single responsibility principle because it now has three responsibilities:

Calling an api
Parsing json
Retrying

This makes it hard to re-use. Let's fix it by being lazy. To keep things general, we'll define a type-alias that models lazy values that are produced with or without side-effects. Let's call this alias Effect since it allows us to treat side-effects as first-class values that can be manipulated by our program, thereby taking the "side" out of "side-effect".

from typing import Callable, TypeVar


A = TypeVar('A')
Effect = Callable[[], A]

We'll use this alias to implement the function retry which can take any Effect and retry it with the same backoff mechanism from before:

def retry(request: Effect[A],
          retry_attempts: int = 3) -> A:
    for attempt in range(1, retry_attempts + 1):
        try:
            return request()
        except Exception as e:
            time.sleep(attempt)
            last_exception = e
    raise last_exception


def get_json_with_retry(url: str, retry_attempts: int = 3) -> dict:
    return retry(lambda: get_json(url), retry_attempts=retry_attempts)

retry treats the results of (for example) HTTP requests as lazy values that can be manipulated. I'm using the term lazy values here because it fits nicely with our theme, but really I could just have said that retry uses get_json as a higher-order function. By treating functions as lazy values that can be passed around, retry achieves more or less total re-usability.

So far so good. Now lets implement a function that executes lazy values in parallel called run_async. Obviously, we want to be able to use run_async with get_json and retry:

from typing import Iterable
from multiprocessing import Pool


def run_async(effects: Iterable[Effect[A]]) -> Iterable[A]:
    with Pool(5) as pool:
        results = [pool.apply_async(effect) for effect in effects]
        return [result.get() for result in results]


def get_json_async_with_retry(urls: Iterable[str], retry_attempts: int = 3) -> Iterable[dict]:
    # lambda url=url is a small "hack" to prevent the url to
    # be mutated in the closures of the lambdas by the for loop
    effects = [lambda url=url: get_json_with_retry(url, retry_attempts) for url in urls]
    return run_async(effects)

(This won't actually work since we have lambdas in the mix and multipocessing doesn't like that without third party libraries, but bear with me.)

That's all fine, but you might object that we've made a mess of our code. I agree. In particular I think that our "glue" functions get_json_with_retry and get_json_async_with_retry are embarrasingly clumsy. What's missing, in my view, is a general solution for glueing lazy values together that would make these specialized glue functions redundant.

To achieve that, we'll use the following dogma:

Functions that perform side effects return Effect instances. In other words, rather than peforming a computation or side-effect directly, they return a lazy description of a result (and/or side-effect) that can be combined with functions that operate on Effect instances.
Functions that operate on Effect instances return new Effect instances

With this scheme, any lazy result can be composed infinitely with functions that operate on lazy results. Indeed, functions that operate on lazy results can be composed with themselves!

So let's maximize the re-usability of our solution by turning our laziness up to The Dude levels of lethargy.

Lets start by refactoring get_json to return an Effect.

def get_json(url: str) -> Effect[dict]:
    def effect() -> dict:
        with urllib.request.urlopen(url) as response:
            content = response.read()
            return json.loads(content)
    return effect

Pretty straight-forward. Now let's do the same for retry and run_async

def retry(request: Effect[A],
          retry_attempts: int = 3) -> Effect[A]:
    def effect() -> A:
        for attempt in range(1, retry_attempts + 1):
            try:
                return request()
            except Exception as e:
                time.sleep(attempt)
                last_exception = e
        raise last_exception
    return effect


def run_async(effects: Iterable[Effect[A]]) -> Effect[Iterable[A]]:
    def effect() -> Iterable[A]:
        with Pool(5) as pool:
            results = [pool.apply_async(effect) for effect in effects]
            return [result.get() for result in results]
    return effect

With this in hand we can compose any variation of our functions to our hearts desire with minimal effort:

url = ...
get_json_with_retry: Effect[dict] = retry(get_json(url))

urls = [...]
get_json_async_with_retry: Effect[Iterable[dict]] = run_async(
    [retry(get_json(url)) for url in urls]
)

First, realise that we could further re-use both get_json_with_retry and get_json_async_with_retry with any function that operates on Effect instances. Also, notice that lazyness (or higher-order functions) is what enables us to program with this degree of re-use, and what allows compositionality on this high level of abstraction (ultimately on the level of entire programs).

When functional programmers claim that programming in functional style makes you more productive, this is the reason: functional programming (which often involves lazy evaluation) can drastically improve re-useability and compositionality, which means you can do much more with much less. All of these advantages were part of my motivation for authoring the library pfun that makes it possible to write Python in functional style without all of the boilerplate and ceremony in this example.

As an added bonus, functional programs are often more predictable and easy to reason about. Also, with few modifications, the pattern we have developed here can be extended to enable completely type-safe dependency injection and error handling. Moreover, static type checking becomes more useful because all functions must return values, even if they merely produce side-effects (in which case they'll return Effect[None]).

In our effort to refactor our example, we have halfway re-invented a common functional pattern: The enigmatic IO type (which we call Effect in our example, and which I hope you don't find mysterious at all at this point). There is much confusion about what IO is and does, especially outside the functional programming clubs of Haskell and Scala programmers. As a consequence you'll sometimes hear IO explained as:

Functions that perform side-effects return IO
Functions that do not perform side-effects do not

While this explanation is not strictly speaking incorrect, it's wildly inadequate because it doesn't mention anything about lazyness which is a core feature of IO. Based on this naive explanation, you might be tempted to try something like:

from typing import TypeVar, Generic


A = TypeVar('A')


class IO(Generic[A]):
    def __init__(self, value: A):
        self.value = value


def get_json(url: str) -> IO[dict]:
    with urllib.request.urlopen(url) as response:
        content = response.read()
        return IO(json.loads(content))

This IO implementation simply tags return values of functions that perform IO, making it clear to the caller that side-effects are involved. Whether this is useful or not is a matter of some controversy, but it doesn't bring any of the functional programming benefits that we have discussed in this article because this IO version is eager.

In summary: lazyness (or higher-order functions) enables radical re-use and compositionality, both of which makes you more productive. To get started with functional programming in Python, checkout the pfun documentation, or the github repository.

suned / pfun

Functional, composable, asynchronous, type-safe Python.

Purely Functional Python With Static Types

Sune Debel — Fri, 03 Jul 2020 20:50:55 +0000

Programming in functional style has never been popular among Python programmers. Even though libraries such as functools or toolz exist, imperative and object-oriented style is still the norm. Being a functional programming convert myself, I'm not sure why that is. I suspect that the lack of a static type system has been a big reason that Python programmers generally haven't recognized the advantages of functional programming, since the marriage between the two is what brings many of the benefits.

With PEP 484 and friends this has changed. The ecosystem around the typing module has matured to the point where a full fledged functional library with static types is practical. As a true functional programming believer this motivated me to author a library called pfun which brings functional programming with a strong emphasis on static types to Python. pfun distinguishes itself from other libraries such as toolz in that

It's designed to provide the most accurate typing possible within the limitations of the type system
It provides a unified functional system for managing side-effects in a module called pfun.effect

This article will explain the pfun effect system in enough detail that you should be able to use it, and maybe even contribute to it. we'll start with an introduction to functional programming and its benefits. If you're already a true believer and functional programming aficionado, you can probably skip ahead to A Functional Effect System for Python.

What is Functional Programming And Should You Care?

In functional programming, we build our programs by combining functions.

But we only use special functions: We only allow functions that compute new results based on their arguments. Functions that do other things besides computing results, such as mutating program state or writing to a file, are not allowed.

We call these "other things" side effects. The reason we don't like functions that perform side effects is that they are harder to re-use since the programmer needs to keep track of which parts of the program (or external state that the program depends on) are affected by the side effect when invoking the function. We prefer functions without side effects to the point that we have named them pure functions, and dislike functions that perform side effects to the extent that we have named them impure functions.

Look at these two functions that both add an element to a list for example:

from typing import List


def add_with_side_effect(l: List[int], e: int) -> List[int]:
    l.append(e)
    return l


def add_without_side_effect(l: List[int], e: int) -> List[int]:
    return l + [e]

Calling add_with_side_effect mutates the list passed as an argument. You need to consider which other parts of the program has a reference to l when calling it. In other words you need to think about your program globally in order to figure out if calling add_with_side_effect is safe.

Calling add_without_side_effect returns a new list. The list the caller has and passes as the argument l is left intact. It doesn't matter if any other part of your program has a reference to l since it's left unchanged. In other words you can reason about your program locally, which is much, much easier most of the time.

Functions that are side-effect-free have a number of significant advantages:

Limiting ourselves to using pure functions also presents a challenge, since most programs in the real world need to read and write files, raise exceptions and use the network. Given that all of these things are inherently impure, it would seem that functional programming is close to useless.

In the next section we'll see how to do all of those things in a purely functional context.

A Functional Effect System for Python

How do we write functions that can handle side-effects and are still pure? To achieve this seemingly impossible feat we're going to use a clever trick. The trick is to change our impure functions such that instead of performing a side effect, they return a description of a side effect that can be executed at a later time. The most straight-forward way of doing that is to write functions that return other functions that performs the side effects. To that end we'll use the following Effect type alias:

from typing import TypeVar, Callable


A = TypeVar('A')
Effect = Callable[[], A]

Lets define a function read_file_pure that returns a description of the side effect of reading a file:

def read_file_pure(path: str) -> Effect[str]:
   def effect() -> str:
       with open(path) as f:
           return f.read()
   return effect

Notice that read_file_pure really is pure: for the same argument path it always returns the same function regardless of the actual contents of the file. The effect function returned by read_file_pure is not pure of course.

So far so good, but how do we actually do anything with the contents of a file? We could of course run the function returned by read_file_pure immediately after calling it to get the content and then perform whatever transformation is necessary - but then our program is no longer pure!

What we'll do instead is to write a new function that takes an Effect and applies a function to its result in a new effect. Lets call this function map (because it maps a function over the result of executing an impure function):

from typing import TypeVar


A = TypeVar('A')
B = TypeVar('B')


def map(effect: Effect[A],
        f: Callable[[A], B]) -> Effect[B]:
    def new_effect() -> B:
        return f(effect())
    return new_effect

Again, notice that map is pure because it is lazy: No side-effects are carried out when calling map. It's only when the function returned by map is called that the side-effect described by the effect parameter is carried out.

This means that we can manipulate values returned by read_file_pure in a pure fashion, for example in:

def lower_case_file(path: str) -> Effect[str]:
    return map(impure_read_file(path),
               lambda contents: contents.lower())

Most functional programs take the shape:

1) Create an Effect to get input for the computation
2) Transform the input using map (or a similar function)
3) Run the Effect returned by map

Or in code:

def transform(input: str) -> str: ...


if __name__ == '__main__'
    program = map(impure_read_file(path), transform)
    _ = program()

In other words, program is an impure function, built by cleverly combining lazy, pure functions. Calling impure_read_file and map doesn't actually do any real computation, except whats required to put together the program function. The real computation, including side-effects, are only executed when program is called at the edge of our python script (in the __main__ section).

That's great, but what if we want to combine several effects? For example, what if we had a function for writing strings to files without side-effects:

def write_file_pure(path: str, contents: str) -> Effect[None]:
    def effect() -> None:
        with open(path, 'w') as f:
            f.write(contents)
    return effect

If we try to use write_file_pure with map, we end up with an Effect[Effect[str]] because map only takes care of "unwrapping" the effect it takes as a parameter, not the effect returned by the function that is mapped over the result of the effect parameter. This means that our program now looks like this:

if __name__ == '__main__':
    program = map(
        read_file_pure('foo.txt'), 
        lambda content: write_file_pure('bar.txt', content)
    )
    _ = program()()  # program() returns an Effect

Notice that we have to call the Effect returned by program() to actually run the effect returned by write_file_pure. Also realize that it's possible to produce an Effect that wraps an arbitrary number of nested effects, for example if map is called in a loop. We might end up in a situation where it's not easy to figure out precisely how many functions we need to call at the edge of our script!

To fix this, we introduce another function that knows how to combine an Effect with a function that returns an Effect. Lets call it and_then since it's chaining together two effects sequentially:

def and_then(effect: Effect[A],
             f: Callable[[A], Effect[B]]) -> Effect[B]:
    def new_effect() -> B:
        return f(effect())()
    return new_effect

Now we can rewrite our program from before as:

if __name__ == '__main__':
    program = and_then(
        read_file_pure('foo.txt'), 
        lambda content: write_file_pure('bar.txt', content)
    )
    _ = program()

Notice that we only have to call program to run all the effects involved in the computation since and_then knows how to combine two effects. Also realize that if we use and_then appropriately, we can combine an arbitrary number of effects, even in a loop. In other words, we have a completely general framework for working with side-effects in a pure fashion:

Functions that need to perform side-effects return Effect instances
We transform the result of effects using map
We combine effects with functions that return effects using and_then

Since functional programs are generally composed of chained calls to map and and_then, lets refactor the functions as instance methods so that we can use a nicer dot notation:

from dataclasses import dataclass
from typing import Callable, TypeVar, Generic


A = TypeVar('A', covariant=True)
B = TypeVar('B')


@dataclass(frozen=True)
class Effect(Generic[A]):
    effect: Callable[[], A]

    def __call__(self):
        return self.effect()

    def map(self, f: Callable[[A], B]) -> Effect[B]:
        return Effect(lambda: f(self()))

    def and_then(self, f: Callable[[A], Effect[B]]) -> IO[B]:
        return Effect(lambda: f(self())())

(We've decorated the class with dataclass to remove some boilerplate and avoid mutation of the effect field). Now our program from before becomes:

def read_file_pure(path: str) -> Effect[str]:
    def effect() -> str:
        with open(path) as f:
            return f.read()
    return Effect(effect)


def write_file_pure(path: str, content: str) -> Effect[None]
    def effect() -> None:
        with open(path, 'w') as f:
            f.write(content)
    return Effect(effect)


if __name__ == '__main__':
    program = read_file_pure(
        'foo.txt'
    ).and_then(
        lambda content: write_file_pure('bar.txt', content)
    )
    program()

Which arguably reads a little nicer.

Adding Error Handling To Our Effect System

So far we have only been concerned with one particular side-effect: IO. Lets now turn our attention to error handling that is also normally modeled as a side-effect in imperative programming.

Our requirement for pure functions is that they can only compute values based on their arguments. To comply with this dogma, lets define wrapper types that can be used to 'tag' values returned from functions.

from dataclasses import dataclass
from typing import Union, TypeVar, Generic

R = TypeVar('R', covariant=True)
L = TypeVar('L', covariant=True)


@dataclass(frozen=True)
class Right(Generic[R]):
    get: R


@dataclass(frozen=True)
class Left(Generic[L]):
    get: L


Either = Union[Left[L], Right[R]]

For convenience, we've also defined a type-alias Either that we can use to more easily write type signatures for functions that can fail. Since it's a union type, it gives us the added benefit that our type checker can help us figure out if we are handling errors correctly.

With this in hand we can do error handling in a purely functional way:

def divide(nume: float, denom: float) -> Either[str, float]:
    try:
        return Right(nume / denom)
    except ZeroDivisionError:
        return Left('Division by zero')

That's great, but what if we need to write a function that needs to model both the IO side-effect and the error handling side-effect? We can of course wrap one in the other, e.g:

def read_file_pure(path: str) -> Effect[Either[IOError, str]]:
    def effect() -> Either[IOError, str]:
        try:        
            with open(path) as f:
                return Right(f.read())
        except IOError as e:
                return Left(e)
    return Effect(action)

Except now we have to deal with error handling every time we need to use map or and_then, because neither of them know how to unpack the value wrapped by an Either:

def lower_case_file(path: str) -> Effect[Either[IOError, str]]:
    read_file_pure(path).map(
        lambda either: Right(either.get.lower()) 
                       if isinstance(either, Right) 
                       else either
    )

Yuck. Thankfully, there are a number of ways to avoid having this error handling code everywhere in our functional program. The solution we use in pfun.effect is to "bake error handling into our Effect type", in order that map and and_then can take care of the wrapping/unwrapping for us:

from dataclasses import dataclass
from typing import Union, TypeVar, Generic


A = TypeVar('A', covariant=True)
E = TypeVar('E', covariant=True)
B = TypeVar('B')
E2 = TypeVar('E2')


@dataclass(frozen=True)
class Effect(Generic[E, A]):
    effect: Callable[[], Either[E, A]]

    def __call__(self) -> Either[E, A]:
        return self.effect()

    def map(self, f: Callable[[A], B]) -> Effect[E, B]:
        def effect():
            either = self()
            if isinstance(either, Left):
                return either
            else:
                return Right(f(either.get))
        return Effect(effect)

    def and_then(self, 
                 f: Callable[[A], Effect[E2, B]]) -> Effect[Union[E, E2], B]:
        def effect():
            either = self()
            if isinstance(either, Left):
                return either
            else:
                return Right(f(either.get)())
        return Effect(effect)

We have added a type parameter E to represent the error type. The implementations of map and and_then are more less identical to our previous versions, except that they expect an Either returned from the wrapped effect function and handle them appropriately. Notice also that the return type of and_then is Effect[Union[E, E2], B], because the resulting Effect can fail if either self() returns a Left value or if f(...)() returns a Left value.

Now the mess from before simply becomes

def read_file_pure(path: str) -> Effect[IOError, str]:
    def effect() -> Either[IOError, str]:
        try:        
            with open(path) as f:
                return Right(f.read())
        except IOError as e:
                return Left(e)
    return Effect(effect)


def lower_case_file(path: str) -> Effect[IOError, str]]:
    read_file_pure(path).map(lambda content: content.lower())

Neato!

Adding The Reader Effect

The Effect type we've cooked up so far is pretty close to what you'll find in pfun.effect. The main difference is that the wrapped effect function in pfun.effect takes a generic argument R:

from dataclasses import dataclass
from typing import TypeVar, Generic, Union


R = TypeVar('R', contravariant=True)
E = TypeVar('E', covariant=True)
E1 = TypeVar('E1') 
A = TypeVar('A', covariant=True)
B = TypeVar('B')


@dataclass(frozen=True)
class Effect(Generic[R, E, A]):
    effect: Callable[[R], Either[E, A]]

    def __call__(self, r: R) -> Either[E, A]:
        return self.effect(r)

    def map(self, f: Callable[[A], B]) -> Effect[R, E, B]:
        def effect(r: R) -> Either[E, B]
            either = self(r)
            if isinstance(either, Left):
                return either
            return Right(f(either.get))
        return Effect(effect)

    def and_then(self, 
                 f: Callable[A, Effect[R, E1, B]]
                ) -> Effect[R, Union[E, E1], B]:
        def effect(r: R) -> Either[E, B]:
            either = self(r)
            if isinstance(either, Left):
                return either
            return Right(f(either.get)(r))
        return Effect(effect)

Allowing the wrapped effect function to take a parameter enables dependency injection for increased code re-use and improved test-ability

For historical reasons, this is known as the reader effect. We will go into what pfun.effect can do with this argument in a lot of detail in future posts, but for now lets just exemplify the dependency injection feature.

Say for example that you want to implement an effect that reads from a database. Using the R parameter you can pass the connection string to the action, making it possible to re-use the same Effect instance for reading from different databases:

def read_from_database(query: str) -> Effect[str, IOError, dict]:
    def effect(connection_string: str) -> Either[IOError, dict]:
        try:
            database = connect(connection_string)
            return Right(database.execute(query))
        except IOError as e:
            return Left(e)
    return Effect(effect)

This allows us to combine read_from_database with other functions through map and and_then, and then re-use the entire function call chain against different databases in a completely type-safe manner.

if __name__ == '__main__':
    program = read_from_database('select * from user').map(lambda users: users['user1'])
    test_results = program('test_user@test_database')
    production_results = program('prod_user@prod_database')

Take a look at the documentation for more information on how to use the R parameter to its full potential.

Whats Next?

There is still a significant problem with the Effect type we have built in this post: long chains of effects combined with map and and_then causes RecursionError because one effect needs to be executed before the next one can be executed.

pfun.effect solves this problem and more:

pfun.effect can build sequences of effects of arbitrary length without causing a RecursionError when executed, because effects are run using a mechanism called trampolines
pfun.effect is compatible with asyncio for high performance asynchronous IO
pfun.effect provides a number of useful helper functions to help you write code at a very high level of abstraction
pfun.effect provides a plugin for mypy that enables fine grained type inference of helper functions

In conclusion, pfun.effect provides a full fledged purely functional, statically typed effect system for Python. Head over to the documentation to go more in depth, or to the github repo for issues, questions and contributions.