Railway Oriented Programming (ROP)#
We don’t really want to raise exceptions since it makes the code bloated with error checking
It’s easy to forget to handle exceptions, or handle the wrong type of exception
Dependencies might even change the kind of exceptions they throw
Let’s model errors using types instead
class Result:
pass
class Ok(Result):
def __init__(self, value):
self._value = value
def __str__(self):
return "Ok %s" % str(self._value)
class Error(Result):
def __init__(self, exn):
self._exn = exn
def __str__(self):
return "Error %s" % str(self._exn)
The Expression library contains a similar but more feature complete Result class we can use:
from expression import Ok, Error
def fetch(url):
try:
if not "http://" in url:
raise Exception("Error: unable to fetch from: '%s'" % url)
value = url.replace("http://", "")
return Ok(value)
except Exception as exn:
return Error(exn)
result = fetch("http://42")
print(result)
Ok 42
def parse(string):
try:
value = float(string)
return Ok(value)
except Exception as exn:
return Error(exn)
result = parse("42")
print(result)
Ok 42.0
Composition#
How should we compose Result returning functions? How can we make a fetch_parse from
fetch and parse.
We cannot use functional composition here since signatures don’t match.
def compose(fn: Callable[[A], Result[B, TError]], gn: Callable[[B], Result[C, TError]]) -> Callable[[A], Result[C, TError]]:
lambda x: ...
First we can try to solve this with an “imperative” implementation using type-checks and
if-else statements:
def fetch_parse(url):
b = fetch(url)
if isinstance(b, Ok):
val_b = b._value # <--- Don't look inside the box!!!
return parse(val_b)
else: # Must be error
return b
result = fetch_parse("http://42")
print(result)
---------------------------------------------------------------------------
TypeError Traceback (most recent call last)
Cell In[6], line 9
6 else: # Must be error
7 return b
----> 9 result = fetch_parse("http://42")
10 print(result)
Cell In[6], line 3, in fetch_parse(url)
1 def fetch_parse(url):
2 b = fetch(url)
----> 3 if isinstance(b, Ok):
4 val_b = b._value # <--- Don't look inside the box!!!
5 return parse(val_b)
TypeError: isinstance() arg 2 must be a type, a tuple of types, or a union
This works, but the code is not easy to read. We have also hard-coded the logic to it’s not possible to easily reuse without copy/paste. Here is a nice example on how to solve this by mixing object-oriented code with functional thinking:
class Ok(Result):
def __init__(self, value):
self._value = value
def bind(self, fn):
return fn(self._value)
def __str__(self):
return "Ok %s" % str(self._value)
class Error(Result):
def __init__(self, exn):
self._exn = exn
def bind(self, fn):
return self
def __str__(self):
return "Error %s" % str(self._exn)
def bind(fn, result):
"""We don't want method chaining in Python."""
return result.bind(fn)
result = bind(parse, fetch("http://42"))
print(result)
def compose(f, g):
return lambda x: f(x).bind(g)
fetch_parse = compose(fetch, parse)
result = fetch_parse("http://123.0")
print(result)
result = fetch("http://invalid").bind(parse)
print(result)
But what if we wanted to call fetch 10 times in a row?#
This is what’s called the “Pyramide of Doom”:
from expression.core import result
result = bind(parse,
bind(lambda x: fetch("http://%s" % x),
bind(lambda x: fetch("http://%s" % x),
bind(lambda x: fetch("http://%s" % x),
bind(lambda x: fetch("http://%s" % x),
bind(lambda x: fetch("http://%s" % x),
bind(lambda x: fetch("http://%s" % x),
fetch("http://123")
)
)
)
)
)
)
)
print(result)
Can we make a more generic compose?#
Let’s try to make a general compose function that composes two result returning functions:
def compose(f, g):
return lambda x: f(x).bind(g)
fetch_parse = compose(fetch, parse)
result = fetch_parse("http://42")
print(result)
Pipelining#
Functional compose of functions that returns wrapped values is called pipeling in the Expression library. Other languages calls this “Kleisli composition”. Using a reducer we can compose any number of functions:
from functools import reduce
def pipeline(*fns):
return reduce(lambda res, fn: lambda x: res(x).bind(fn), fns)
Now, make fetch_and_parse using kleisli:
fetch_and_parse = pipeline(fetch, parse)
result = fetch_and_parse("http://123")
print(result)
What if we wanted to call fetch 10 times in a row?#
from expression.extra.result import pipeline
fetch_with_value = lambda x: fetch("http://%s" % x)
request = pipeline(
fetch,
fetch_with_value,
fetch_with_value,
fetch_with_value,
fetch_with_value,
fetch_with_value,
fetch_with_value,
parse
)
result = request("http://123")
print(result)
Result in Expression#
The Result[T, TError] type in Expression lets you write error-tolerant code that can
be composed. A Result works similar to Option, but lets you define the value used for
errors, e.g., an exception type or similar. This is great when you want to know why some
operation failed (not just Nothing). This type serves the same purpose of an Either
type where Left is used for the error condition and Right for a success value.
from expression import effect, Ok, Result
@effect.result[int, Exception]()
def fn():
x = yield from Ok(42)
y = yield from Ok(10)
return x + y
xs = fn()
assert isinstance(xs, Result)
A simplified type called Try is also available. It’s a result type
that is pinned to Exception i.e., Result[TSource, Exception]. This makes the code
simpler since you don’t have specify the error type every time you declare the type of
your result.