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.