Using Either monad in Python

Roman Nesytov
3 min readMay 3, 2018

--

I can’t write spaghetti code anymore. Konstantin Flavitsky 1864 Oil on canvas

Introduction

There are many abstractions in programming. Some seem easy to understand, some are not. Let’s look on monads.

Some developers burn at the stake code with monads using. The main reason— certainty that they need to know category theory, Haskell and got a Ph.D. Other developers are certain that use of this abstraction in Python is ridiculous.

The main goal of this text is to show by examples the use of monads in real Python code, rather than explain monads.

The code

Most of our code is work with data. We receive data, we validate it, we process it and we write it. Example:

def perform_some_operation(url):
reponse = http.get(url)

if response.status == 200:
validation_result = validator(reponse.body)

if validation_result.is_valid():
try:
user = User.objects.get(response.body['id'])

call_notification.delay(user=user)
except User.DoesNotExist:
return 'User does not exists'
else:
return 'Invalid response data'
else:
return 'Invalid response status'

This code looks like spaghetti with meatballs. It looks ugly, not enough readable and testable. Let’s look how Either monad can improve this code.

Either is a facility for dealing with operations that can fail. It instance represents an abstract type that may be in one of two possible states, called a Left and a Right. By convention, the Right state is the proper (boxed) value and the Left state is used for diagnostics.

OSlash implements monads in Python pretty well.

>>> from oslash import Left, Right
>>> right = Right(1)
>>> right.value
1
>>> left = Left(1)
>>> left.value
1

Left and Right just store value which get by constructor. The main difference is bind method implementation.

class Right(Either):
def __init__(self, value):
self._value = value

def bind(self, func):
return func(self._value)
class Left(Either):
def __init__(self, value):
self._value = value

def bind(self, func):
return Left(self._value)

Right applies func to it value and Left returns itself.

>>> Right(1).bind(lambda x: x + 1)
2
>>> Left(1).bind(lambda x: x + 1)
<oslash.either.Left object at 0x1076ae4a8>
>>> Left(1).bind(lambda x: x + 1).value
1

So, if func returns other Left or Right instance, then we can chain calls!

>>> Right(5) \
... .bind(lambda x: Right(x + 1)) \
... .bind(lambda x: Right(x * 2)).value
12

But if we replace Right(5) with Left(5).

>>> Left(5) \
... .bind(lambda x: Right(x + 1)) \
... .bind(lambda x: Right(x * 2)).value
5

OSlash redefines __or__ method.

class Monad():
...
def __or__(self, func):
return self.bind(func)
...

So we can chain call using | operator.

>>> result = Right(5) | (lambda x: Right(x + 1)) | (lambda x: Right(x * 2))
>>> result.value
12

We can use Either to build something like pipelines. If some step of pipeline returns Left instance, then other steps will not be executed. Let's rewrite our first example with Either and encapsulate it in service object.

class SomeService():
def make_request(self, context):
reponse = http.get(context['url'])

if response.status == 200:
return Right(dict(context, reponse=response))
else:
return Left('Invalid response')

def validate_response(self, context):
validation_result = validator(reponse.body)

if validation_result.is_valid():
return Right(context)
else:
return Left('Invalid response data')

def find_user(self, context):
response = context['response']

try:
user = User.objects.get(response.body['id'])

return Right(dict(context, user=user))
except User.DoesNotExist:
return Left('User does not exists')

def call_notification(self, context):
call_notification.delay(user=context['user'])

return Right(context)

def __call__(self, url):
Right({'url': url}) | \
self.make_request | \
self.validate_response | \
self.find_user | \
self.call_notification

All steps in shares it data by context, which is just a dictionary. Any step which finished with success should return Right with context as value.

Now our service return either Left or Right instance depending on the success of pipeline processing. We can check result using isinstance.

result = SomeService()('http://example.com')

if isinstance(result, Right):
logger.info('Success')
else:
logger.error('Failed')

This approach makes our code more flexible (we can just add new step by one line) and testable (we can mock any step).

--

--