Monadic computations in Python

Monads

Monads are a construct known from category theory. Newcomers to Haskell quickly connect the term with mixed feelings, because the concept is rather strange in the beginning, especially if you already know another ‘traditional’ language, but heavily used within the Haskell programming language. The first contact with them is usually when doing IO in combination with do-notation.

Why are monads used so much in Haskell? Because they allow to sequence computations in a functional manner. In addition monads can hide behavioral boilerplate within their definitions: Define the behavior once for the monad instance and get it ‘for free’ with do-notation.

Once the concept of a monad is understood the next level of insight are so called monad transformers. Monad transformers are monads stacked atop a base monad, i.e. monad stacks. Every stack layer adds another piece of functionality and the final resulting monad posses all of these.

This post is not meant as an introduction to monads. Many excellent books cover this topic and yet another post about monads does not add any good. However the benefit of monads is not limited to the Haskell programming language. In this post I want to show how monads can be used in Python.

Note: The code samples in this post require at least Python v3.6.

Python, IO and Monads

Python is an imperative language and thus fairly different from Haskell, which is considered a functional language. Whilst IO is explicit in Haskell it can occur anywhere within Python source code. Monads are not necessary to sequence computations in Python, statements and expressions are simply executed from top to bottom. Another difference is that Haskell is a strongly-typed compiled language whereas Python is a dynamic interpreted language. Thus from a language theory point of view both languages are situated in completely opposite corners of the scale.

However even though you don’t need the concept of a monad as a Python programmer, doesn’t mean that you can’t benefit from the ideas of monadic computations to represent behavior. In the following sections I want to start with a real world scenario and a rather typical approach to the problem and then evolve step by step to a monadic solution.

Note: There is no single truth about when a programming language is considered a functional programming language. For an interesting discussion about this see this post of Michael Snoyman. However the important thing is: even imperative languages allow for a functional style of programming. The main differences are just to what extent this kind of style is supported by the language’s syntax, semantics and the standard library.

To implement monads in Python I will use the following concepts:

  • Closures
  • Lambda functions
  • Higher order functions
  • Operator overloading

The Challenge

Let me introduce the task we try to solve. The example is taken from the real world and is actually something I had to do in my job recently.

We want to control an external program with a Python script. A very typical example for such a program is git. Git is an amazing version control system and I use it a lot at work. There are a lot of graphical tools to support working with git but my favorite is the plain old terminal interface. However working with git requires to know a lot of different commands and to execute them in a certain order and potentially every single command can fail. Git follows the UNIX convention that a return code greater than zero signals failure.

In continuous integration scenarios you usually have to execute a lot of these git commands and of course you don’t want to continue if the previous command already failed. Likewise you want to know what has been done, in the good and bad case, so standard out and error should be captured.

So let’s sum up the requirements:

  1. We want to call an external program from within our Python code
  2. Every invocation of the program can potentially fail
  3. If a command fails we don’t want to continue, but instead give a meaningful error message.
  4. We want to capture standard out and standard error of the program

Note: In the following I will not actually use git commands, but rather invoke some dummy shell scripts that either exit with zero or some value greater than zero, but this does not harm the generality of the following code snippets.

Imperative Solution

This solution will serve as the reference solution. Of course there are many ways to fulfill the above requirements but I think the solution I give is rather realistic and pythonic.

Brute Force Solution

The run function of the subprocess module takes a check parameter. If this is set to True the return value of the process is checked and an exception is thrown in case it signals error. Thus the probably simplest method is to simply put everything in a try/except block.

from subprocess import run, PIPE, STDOUT


def shell(cmd):
    print(f"Now running: {cmd}")
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True, check=True)
    print(process.stdout.decode('utf-8'))

def main():
    try:
        shell("echo 'OK'; exit 0")
        shell("echo 'Again OK'; exit 0")
        shell("echo 'Problem!'; exit 1")
        shell("echo 'Should not execute'; exit 0")
    except:
        print("An error occurred. Aborting")
    else:
        print("Done")

if __name__ == '__main__': main()
$ python naive.py
Now running: echo 'OK'; exit 0
OK

Now running: echo 'Again OK'; exit 0
Again OK

Now running: echo 'Problem!'; exit 1
An error occurred. Aborting

However raising exceptions is definitely not a good programming praxis and you have to make sure to catch them, because exceptions will bubble up! This is maybe ok if the users are programmers, but it is very ugly if an end user is confronted with a bunch of exception and stack trace outputs instead of a comprehensible error messages.

Naive Solution

As the next iteration to the problem the shell function will return a boolean signaling sucess/failure and the main code will check every single invocation.

from subprocess import run, PIPE, STDOUT


def shell(cmd):
    print(f"Now running: {cmd}")
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    print(process.stdout.decode('utf-8'))
    return process.returncode == 0

def main():
    ok = shell("echo 'OK'; exit 0")
    if not ok:
        print("An error occurred. Aborting")
        return
    ok = shell("echo 'Again OK'; exit 0")
    if not ok:
        print("An error occurred. Aborting")
        return
    ok = shell("echo 'Problem!'; exit 1")
    if not ok:
        print("An error occurred. Aborting")
        return
    ok = shell("echo 'Should not execute'; exit 0")
    if not ok:
        print("An error occurred. Aborting")
        return
    print("Done")

if __name__ == '__main__': main()

Notice how repetitive this feels and looks. What’s even worse: It’s hard to see what is the program actually trying to do! Sometimes you also see something like this:

from subprocess import run, PIPE, STDOUT


def shell(cmd):
    print(f"Now running: {cmd}")
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    print(process.stdout.decode('utf-8'))
    if process.returncode == 0:
        return True
    else:
        print("An error occurred. Aborting")
        return False

def main():
    ok = shell("echo 'OK'; exit 0")
    if ok:
        ok = shell("echo 'Again OK'; exit 0")
        if ok:
            ok = shell("echo 'Problem!'; exit 1")
            if ok:
                ok = shell("echo 'Should not execute'; exit 0")
                if ok:
                    print("Done")

if __name__ == '__main__': main()

This is slightly more readable, but the constantly growing indentation depth scales badly and notice that the error message is now baked into the shell function, which cuts down on reusability.

Better Solution - Higher Order Functions

Using higher order functions (functions taking functions as arguments) we can feed our list of functions (note we do not execute them at that point!) to a driver function. The driver then executes the functions one after another until either one of them fails or the end is reached.

from subprocess import run, PIPE, STDOUT


def do_shell(cmd):
    print(f"Now running: {cmd}")
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    print(process.stdout.decode('utf-8'))
    return process.returncode == 0

def shell(cmd):
    return lambda: do_shell(cmd)

def run_in_sequence(cmds):
    for cmd in cmds:
        ok = cmd()
        if not ok:
            return False
    return True

def main():
    ok = run_in_sequence([
           shell("echo 'OK'; exit 0"),
           shell("echo 'Again OK'; exit 0"),
           shell("echo 'Problem!'; exit 1"),
           shell("echo 'Should not execute'; exit 0")
           ])
    if ok:
        print("Done")
    else:
        print("An error occurred. Aborting")

if __name__ == '__main__': main()

This solution is a lot more readable and avoids code duplication. The shell function produces a delayed computation (aka thunk) of the desired shell invocation using a lambda function. The underlying do_shell function is kept general, simple and can be easily reused.

There is a problem though. We are just printing the contents of standard out. For simple scripts that’s often all we need. But what if we want to write the contents to a file? Or filter the error messages? We can’t.

Or maybe we don’t want to always print out everything. In this case we have some sort of verbosity level and we only use the print statement if the verbosity is high enough. This will further clutter our code with if/else statements.

Final Solution - Collecting stdout

As the final solution we will collect the stdout values and return a list of them. This gives maximum control over the contents and allows things like filtering or verbosity to be handled in the main function instead of the low level code.

from subprocess import run, PIPE, STDOUT


def do_shell(cmd):
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    return (process.returncode == 0, [f"Now running: {cmd}", process.stdout.decode('utf-8')])

def shell(cmd):
    return lambda: do_shell(cmd)

def run_in_sequence(cmds):
    stdouts = []
    for cmd in cmds:
        ok, stdout = cmd()
        stdouts += stdout
        if not ok:
            return (False, stdouts)
    return (True, stdouts)

def main():
    verbosity = 2
    ok, stdouts = run_in_sequence([
        shell("echo 'OK'; exit 0"),
        shell("echo 'Again OK'; exit 0"),
        shell("echo 'Problem!'; exit 1"),
        shell("echo 'Should not execute'; exit 0")
        ])
    interesting_stuff = list(
                            filter(lambda s: not s.startswith("Now running"),
                                   stdouts))
    if ok and verbosity > 1:
        print("STDOUTS")
        for s in interesting_stuff:
            print(s)
        print("Done")
    else:
        print("An error occurred. Aborting")
        print("STDOUTS")
        for s in interesting_stuff:
            print(s)

if __name__ == '__main__': main()

Most people don’t realize that this final solution is basically a hand-written monadic computation. We will see more of this in the coming sections.

Monadic Solution

Before I dive into the final section I want to spend a few words to recapitulate the involved monads and how they work. Our final solution will be an EitherT transformer stacked on top of a WriterT transformer stacked on top of the IO monad.

The Either Monad

An invocation of our program can either succeed or fail. This either or behavior is implemented by the either monad. In Haskell the either monad, or the either type class, has to constructors, called Left and Right. By convention Left signals an error and Right signals success.

data  Either a b  =  Left a | Right b

-- example for a left value
let leftEx = Left "Error"
-- example for a right value
let rightEx = Right 42

The interesting part is the monad instance of Either:

instance Monad (Either e) where
    return = Right
    Left  l >>= _ = Left l
    Right r >>= k = k r

In short: Whenever we see a Left value in the chain we ignore everything following that and instead return the Left value as the result of the whole computation:

#!/usr/bin/env stack
{- stack script --resolver lts-11.8 -}

ok_computation :: Either String Int
ok_computation = Right(1) >> Right(2) >> Right(3) >> Right(4)

nok_computation :: Either String Int
nok_computation = Right(1) >> Right(2) >> Left("Something bad happened") >> Right(4)

main = do
  print ok_computation
  print nok_computation
$ ./either_demo.hs
Right 4
Left "Something bad happened"

The Writer Monad

We want to capture the content of standard out and standard error for every invocation of the external program. This means we want to persist (write) the content somewhere instead of loosing it. In Haskell the monad responsible for that is the writer monad.

newtype Writer w m a = Writer { runWriter :: (a, w) }

-- example for a writer value
let writerEx = Writer (1,"String with information")

instance (Monoid w) => Monad (Writer w) where
    return x = Writer (x, mempty)
    (Writer (x,v)) >>= f = let (Writer (y, v')) = f x in Writer (y, v `mappend` v')

In short: The first element of the tuple is used to drive the computation as the input to the rest of the computation while the second element accumulates the information value.

#!/usr/bin/env stack
{- stack script
   --resolver lts-11.8
   --package transformers
-}

import Control.Monad.Trans.Writer.Lazy

writer_computation :: Writer String Int
writer_computation = do
  writer (1,"Hello")
  writer (2," World")
  writer (3,"!!!")

main = do
  print $ runWriter writer_computation
$ ./writer_demo.hs
(3,"Hello World!!!")

The IO Monad

As the name says the IO monad is responsible for Input and Output, i.e. the bi-directional communication with “the outside world”. In Haskell the definition of the IO monad is opaque, it is not possible to unwrap IO values except from within the IO monad. This ensures the encapsulation and the partition between pure and impure code that Haskell is famous for.

#!/usr/bin/env stack
{- stack script
   --resolver lts-11.8
   --package base
-}

import Data.Char (toUpper)

echo_shout :: IO ()
echo_shout = do
  line <- getLine
  putStrLn $ (map toUpper line) ++ "!"

main = echo_shout
$ ./io_demo.hs
hi
HI!

Rebuilding Monads in Python

Note: There already exists a Python library that provides implementations for Functor, Applicative and Monad called OSlash. I was not fully convinced by their implementation. It felt a tiny bit to OO for my taste and and I had the impression that it is very hard to implement monad transformers on top of IO with their code. What follows below is my proto-typical code to implement monadic stacks in Python, but my implementation is strongly influenced by OSlash, e.g. the operator overloading.

Either in Python

class Either():
    Left = 0
    Right = 1
    def __init__(self, kind, value=None):
        self._kind = kind
        self._value = value

    def __or__(self, k):
        if isLeft(self):
            return self
        else:
            return k(self.unwrap())

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def unwrap(self):
        return self._value

    def __str__(self):
        if self._kind == Either.Left:
            return f"Left {repr(self._value)}"
        else:
            return f"Right {repr(self._value)}"

    def __repr__(self):
        return self.__str__()


def isLeft(e):
    return e._kind == Either.Left


def isRight(e):
    return e._kind == Either.Right


def left(a):
    return Either(Either.Left, a)


def right(a):
    return Either(Either.Right, a)

# Now we can write code like in the example above
print(right(1) >> right(2) >> right(3) >> right(4))
print(right(1) >> right(2) >> left("Something bad happened") >> right(4))
print(right(1) | (lambda x: right(2*x)))
print(left("nok") | (lambda x: right(2*x)))
$ python either_demo.py
Right 4
Left 'Something bad happened'
Right 2
Left 'nok'

This example uses operator overloading for classes to support a nicer to read infix operator syntax very similar to Haskell. So basically the only reason that we are using classes here is for nicer syntax (operator overloading) and to store information if we have a Left or a Right value.

IO in Python

An IO action in Python is just a deferred call to a function that does IO. Like in Haskell we don’t want IO actions to do anything until they are actually required (run). This mimics the behavior of lazy evaluation.

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

    @staticmethod
    def mreturn(value):
        return IO(lambda: value)

    def unwrap(self):
        return self._value

    def run(self):
        return self._value()

    def __or__(self, k):
        return IO(lambda: k(self.run()).run())

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def __str__(self):
        return "IO {repr(self._value)}"

    def __repr__(self):
        return self.__str__()

io1 = IO.mreturn(1)
def io2(x):
    print(f"Doubling {x} -> {2*x}")
    return IO.mreturn(None)

composite_io = io1 | io2
composite_io.run()
$ python io_demo.py
Doubling 1 -> 2

Monad Transformers in Python

As I mentioned above we want the functionality of Either and Writer. Thus we have to stack these atop our IO monad. Before I go into details, here is the full code:

from functools import reduce
from subprocess import run, PIPE, STDOUT

class Either():
    Left = 0
    Right = 1
    def __init__(self, kind, value=None):
        self._kind = kind
        self._value = value

    def unwrap(self):
        return self._value

    @staticmethod
    def mreturn(value):
        return Either(Either.Right, value)

    def get_return(self):
        return Either.mreturn

    def __or__(self, k):
        if isLeft(self):
            return self
        else:
            return k(self.unwrap())

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def __str__(self):
        if self._kind == Either.Left:
            return f"Left {repr(self._value)}"
        else:
            return f"Right {repr(self._value)}"

    def __repr__(self):
        return self.__str__()


def isLeft(e):
    return e._kind == Either.Left


def isRight(e):
    return e._kind == Either.Right


def left(a):
    return Either(Either.Left, a)


def right(a):
    return Either(Either.Right, a)


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

    @staticmethod
    def mreturn(value):
        return IO(lambda: value)

    def get_return(self):
        return IO.mreturn

    def unwrap(self):
        return self._value

    def run(self):
        return self._value()

    def __or__(self, k):
        return IO(lambda: k(self.run()).run())

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def __str__(self):
        return "IO {repr(self._value)}"

    def __repr__(self):
        return self.__str__()


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

    def get_return(self):
        inner_return = self.unwrap().get_return()
        return lambda value: WriterT(inner_return((value,[])))

    def unwrap(self):
        return self._value

    def __or__(self, k):
        inner_ret = self.unwrap().mreturn
        return WriterT(
            self.unwrap() |(
            lambda x1: k(x1[0]).unwrap() |(
            lambda x2: inner_ret((x2[0], x1[1] + x2[1])))))

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def __str__(self):
        return f"WriterT {repr(self._value)} "

    def __repr__(self):
        return self.__str__()


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

    def get_return(self):
        inner_return = self.unwrap().get_return()
        return lambda value: EitherT(inner_return(right(value)))

    def unwrap(self):
        return self._value

    def __or__(self, k):
        inner_ret = self.unwrap().get_return()
        return EitherT(
            self.unwrap() |(
            lambda a: inner_ret(a) if isLeft(a) else k(a.unwrap()).unwrap()))

    def __rshift__(self, k):
        return self.__or__(lambda _: k)

    def __str__(self):
        return f"EitherT {repr(self._value)} "

    def __repr__(self):
        return self.__str__()


# The low level function handling the execution of the external program.
# In Haskell it would have the signature
# do_shell :: String -> IO (Either () (), [String])
def do_shell(cmd):
    process = run(cmd, stdout=PIPE, stderr=PIPE, shell=True)
    stdout = process.stdout.decode('utf-8')
    stderr = process.stderr.decode('utf-8')
    info = [f"Command run: {cmd}"]
    if stdout:
        info.append(stdout)
    if stderr:
        info.append(stderr)
    if process.returncode > 0:
        return (left(stderr), info)
    return (right(stdout), info)

# Helper function to construct the underlying data type
def shell(cmd):
    # This is the big boy: A monadic stack, stacking WriterT and EitherT on
    # top of the IO monad First the IO action, it must already follow the
    # conventions of the stack, i.e. return a tuple of (Either a b,[String]).
    # The first entry is used by the stacked EitherT monad and the second
    # one is the monoid instance for the stacked WriterT monad
    return EitherT(WriterT(IO(lambda: do_shell(cmd))))


def runAction(action):
    return action.unwrap().unwrap().run()


if __name__ == '__main__':

    # EXAMPLES
    cmd1 = shell("echo 'OK'; exit 0")
    cmd2 = shell("echo 'Also OK'; exit 0")
    cmd3 = shell("echo 'Even better'; exit 0")
    full_action = cmd1 >> cmd2 >> cmd3
    res, info = runAction(full_action)
    print(f"Final result: {res}")
    print("== INFO ==")
    print(reduce(lambda x, y: x + '\n' + y, info))
    cmd4 = shell("echo 'Command failed' >&2; exit 1")
    full_action = cmd1 >> cmd4 >> cmd3
    res, info = runAction(full_action)
    print(f"Final result: {res}")
    print("== INFO ==")
    print(reduce(lambda x, y: x + '\n' + y, info))
$ python3.6 transformers.py
Final result: Right 'Even better\n'
== INFO ==
Command run: echo 'OK'; exit 0
OK

Command run: echo 'Also OK'; exit 0
Also OK

Command run: echo 'Even better'; exit 0
Even better

Final result: Left 'Command failed\n'
== INFO ==
Command run: echo 'OK'; exit 0
OK

Command run: echo 'Command failed' >&2; exit 1
Command failed

As can be seen from the examples the command sequence short-circuits in case one of the commands fails. The implementations for EitherT and WriterT are as close as possible to that of Haskell’s WriterT nad EitherT. The implementation of __or__ (|) corresponds to (>>=) in Haskell and that of __rhift__ (>>) corresponds to Haskell’s (>>). As in Haskell we have to unwrap the wrapped IO action before we can run it. The helper function runAction does the job for us.

There exist a few technical details that are necessary to make this code run under Python:

  • Because Python is not typed, we have to use one of the wrapped inner monadic values to retrieve the return function (constructor) of the inner monad of the transformer, e.g. self.unwrap().get_return()
  • When using the bind operation we have to enclose the following lambda expression in parenthesis so the code can be correctly lexed and parsed
  • Python does not have do-notation, so no syntactic sugar
  • In Haskell unwrapping the value of monads is done with descriptive names, e.g. runWriterT or runEitherT, I use the generic unwrap function for this. In that way I don’t have to deal with type information, because I assume that every monad class has that method

Comparison to the hand-written Code

If we carfully reconstruct the logic added with every monad transformer level we can see that the monadic solution is actually equivalent to the hand-written code at the beginning of the post. At least in the way we used it in the examples.

In the hand-written code we used boolean values to signal success or failure. The Either type is more powerful than that, because it can also store content, but we haven’t used it here. We simply returned None, which then carries the equally much information as the boolean values (Left is False and Right is True).

The monadic code saves the continuation within a closure, the hand-written code uses an explicit for loop.

The information from stdout is kept in a global variable within function scope in case of the hand-written solution, whereas the monadic solution explicitly hands over the updated value to the next function call.

So why bother with all this extra code when the hand-written solution is so similar in functionality? There are a couple of reasons.

Reusability

The IO, Either, Writer, EitherT and WriterT classes are self-contained and thus fully reusable. It is possible to build other monadic stacks with them. This is not true for the hand-written solution.

Composability

Composability is an interesting aspect and shouldn’t be neglected! The sample code I presented is very basic, but a full featured continuous integration script for a complex project might grow to considerable size. It is very likely that the overall command sequences are made of smaller blocks which are frequently repeating. In this case it would be nice to be able to compose the higher level command blocks of the smaller ‘unit’ command blocks.

Example

We want to retrospectively tag a bunch of commits in a git repository and also print information about the commit:

from transformers import *

def create_tag(refspec, tagname, message):
    return (shell(f"git show {refspec}") >>
            shell(f"git tag -m '{message}' {tagname} {refspec}"))

tagAction = (create_tag('df40cfe', 'v1.0', 'First release') >>
             create_tag('851a93b', 'v2.0', 'Second release') >>
             # Ups I made an error - tag already exists
             create_tag('961090f', 'v1.0', 'Third release') >>
             create_tag('32d3085', 'v4.0', 'Forth release')
            )

res, info = runAction(tagAction)
print(f"Final result: {res}")
print("== INFO ==")
for i in info:
    print(i)
$ cd example_git_repo
$ python ../git_tagging_example.py
Final result: Left None
== INFO ==
Command run: git show df40cfe
commit df40cfe4876c9bf57254572762fa9fa669ce0ef3
Author: TobiasPleyer <tobi.pleyer@gmail.com>
Date:   Sun Dec 30 10:57:58 2018 +0100

    First commit

diff --git a/README b/README
new file mode 100644
index 0000000..3609f20
--- /dev/null
+++ b/README
@@ -0,0 +1 @@
+One

Command run: git tag -m 'First release' v1.0 df40cfe

Command run: git show 851a93b
commit 851a93b45397a12358ba697108e05d88928e6584
Author: TobiasPleyer <tobi.pleyer@gmail.com>
Date:   Sun Dec 30 10:58:35 2018 +0100

    Second commit

diff --git a/README b/README
index 3609f20..3b0086f 100644
--- a/README
+++ b/README
@@ -1 +1 @@
-One
+Two

Command run: git tag -m 'Second release' v2.0 851a93b

Command run: git show 961090f
commit 961090f6a727b9c368a1913a8136245aad1d55ee
Author: TobiasPleyer <tobi.pleyer@gmail.com>
Date:   Sun Dec 30 10:59:18 2018 +0100

    Third commit

diff --git a/README b/README
index 3b0086f..b2cde18 100644
--- a/README
+++ b/README
@@ -1 +1 @@
-Two
+Three

Command run: git tag -m 'Third release' v1.0 961090f
fatal: tag 'v1.0' already exists

The git show/git tag block forms the unit and we chain several of these units together to form a bigger block. This will just result in a bigger monadic computation that retains the same behavior as if defined as one big block. That’s the beauty of composability.

But this can also be achieved in the hand-written solution. If we define an action as the list of shell command lambdas then composition can be done via list addition:

from subprocess import run, PIPE, STDOUT


def do_shell(cmd):
    process = run(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    return (process.returncode == 0, [f"Now running: {cmd}", process.stdout.decode('utf-8')])

def shell(cmd):
    return lambda: do_shell(cmd)

def run_in_sequence(cmds):
    stdouts = []
    for cmd in cmds:
        ok, stdout = cmd()
        stdouts += stdout
        if not ok:
            return (False, stdouts)
    return (True, stdouts)

def create_tag(refspec, tagname, message):
    return [shell(f"git show {refspec}"),
            shell(f"git tag -m '{message}' {tagname} {refspec}")]

def main():
    tagAction = (create_tag('df40cfe', 'v1.0', 'First release') +
                 create_tag('851a93b', 'v2.0', 'Second release') +
                 # Ups I made an error - tag already exists
                 create_tag('961090f', 'v1.0', 'Third release') +
                 create_tag('32d3085', 'v4.0', 'Forth release')
                )
    ok, stdouts = run_in_sequence(tagAction)
    print(f"Final result: {ok}")
    print("== INFO ==")
    for s in stdouts:
        print(s)

if __name__ == '__main__': main()

But once again: This shouldn’t surprise us because as I mentioned above the logic in run_in_sequence is basically identical to our monadic code in case we are simple sequencing computations.

Branching

Branching means doing different things, i.e. taking different code execution paths depending on conditionals. Typically the execution of an action influences the behavior or execution of the following actions.

This is where the monadic computations really start to shine. So far we have only looked at blindly sequencing commands after each other with the (>>) operator. This is because the commands were independent of each other. But what if we want to do different things depending on the output of previous commands? In Haskell we would use the (>>=) function, in our Python code we use the (|) operator for that.

Example

We want to execute different commands depending on the branch we are currently checked out to.

from transformers import *


def act_on_branch(branch):
    if branch == 'master':
        return shell("echo 'action for master'")
    else:
        return shell("echo 'Other action'")

def stackReturn(value):
    return EitherT(WriterT(IO.mreturn((right(value),[]))))

action = shell('git branch') |(
         # Extract the branch name from the git output
         lambda bOut: stackReturn(bOut.strip().split()[-1]) |(
         # Behave different depending on branch
         lambda branch: act_on_branch(branch) >>
         shell("echo 'command independent of previous commands'") >>
         # The 'branch' parameter remains in scope
         shell(f"echo {branch}") >>
         shell('git tag | wc -l') |(
         # Convert the string output to an integer
         lambda cnt: stackReturn(int(cnt)) |(
         lambda n: shell(f"/bin/zsh -c 'for i in {{1..{n}}}; do; echo 'Branch!'; done'")
         ))))

res, info = runAction(action)
print(f"Final result: {res}")
print("== INFO ==")
print(reduce(lambda x, y: x + '\n' + y, info))
$ cd example_git_repo
$ python ../branching_example.py
Final result: Right 3
== INFO ==
Command run: git branch
* master

Command run: echo 'action for master'
action for master

Command run: echo 'command independent of previous commands'
command independent of previous commands

Command run: echo master
master

Command run: git tag | wc -l
3

Command run: /bin/zsh -c 'for i in {1..3}; do; echo 'Branch!'; done'
Branch!
Branch!
Branch!

$ git tag newTag
$ git checkout -b other_branch
Switched to a new branch 'other_branch'
$ python ../branching_example.py
Final result: Right 4
== INFO ==
Command run: git branch
  master
* other_branch

Command run: echo 'Other action'
Other action

Command run: echo 'command independent of previous commands'
command independent of previous commands

Command run: echo other_branch
other_branch

Command run: git tag | wc -l
4

Command run: /bin/zsh -c 'for i in {1..4}; do; echo 'Branch!'; done'
Branch!
Branch!
Branch!
Branch!

$ cd /home
$ python branching_example.py
Final result: Left ''
== INFO ==
Command run: git branch
fatal: Not a git repository (or any parent up to mount point /home)
Stopping at filesystem boundary (GIT_DISCOVERY_ACROSS_FILESYSTEM not set).

As the example shows the monad’s bind syntax can be used to change the code execution path easily and readable plus we still retain the nice and controlled behavior in case of an error. The variables bound by lambdas will remain visible to all follow up actions! If more sophisticated decision making is required that can’t be done with the limited lambda syntax in Python we can simply define a helper function to do the job.

Challenge: Try to reproduce that kind of behavior with a hand-written solution!

Some people would argue the above code samples look ugly and that the used lambda function syntax is not very understandable and readable. I can’t deny that, but keep in mind that this is due to the lack of proper syntactic sugar support in Python! Here is the same example written in Haskell:

import Control.Monad.IO.Class
import Control.Monad.Trans.Class
import Control.Monad.Trans.Either
import Control.Monad.Trans.Writer.Lazy
import Control.Monad (liftM)
import Data.String.Utils
import System.Exit
import System.IO
import System.Process hiding (shell)


type MyStack a = EitherT String (WriterT [String] IO) a


shell :: String -> MyStack String
shell cmd = do
  (exitCode, stdout, stderr) <- liftIO $ readProcessWithExitCode "/bin/zsh" ["-c", cmd] []
  if (exitCode == ExitSuccess)
  then EitherT $ writer (Right (stdout), ["Executing '" ++ cmd ++ "'", stdout])
  else EitherT $ writer (Left (stdout), ["Executing '" ++ cmd ++ "'", stdout, stderr])

act_on_branch b =
  if (b == "master")
  then shell "echo 'action for master'"
  else shell "echo 'Other branch'"

stackReturn value = return $ return $ return (Right value, [])

main = do
  (res,info) <- runWriterT $ runEitherT $ do
    bOut <- shell "git branch"
    let branch = last $ words bOut
    act_on_branch branch
    shell "echo 'command independent of previous commands'"
    shell $ "echo " ++ branch
    cnt <- strip <$> shell "git tag | wc -l"
    shell $ "for i in {1.." ++ cnt ++ "}; do; echo 'Branch!'; done"
  case res of
    Right _ -> putStrLn "SUCCESS"
    Left _ -> putStrLn "FAILURE"
  print res
  putStrLn "== INFO =="
  mapM_ putStrLn info

Summary

Even if you’re not convinced at that point that monadic computations can help you in everyday programming I think it is absolutely worth it to at least know the bigger picture behind the above shown mechanics. The concept of a monad is an extremely versatile and powerful one which allows for arbitrarily complex computations. Haskell’s IO system is the proof that anything can be done with monads.