Example Usage

To show how Mush works from a more practical point of view, let’s start by looking at a simple script that covers several common patterns:

from argparse import ArgumentParser
from .configparser import RawConfigParser
import logging, os, sqlite3, sys

log = logging.getLogger()

def main():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('--quiet', action='store_true',
                        help='Log less to the console')
    parser.add_argument('--verbose', action='store_true',
                        help='Log more to the console')
    parser.add_argument('path', help='Path to the file to process')

    args = parser.parse_args()
    
    config = RawConfigParser()
    config.read(args.config)
    
    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log.addHandler(handler)
    log.setLevel(logging.DEBUG)

    if not args.quiet:
        handler = logging.StreamHandler(sys.stderr)
        handler.setLevel(logging.DEBUG if args.verbose else logging.INFO)
        log.addHandler(handler)

    conn = sqlite3.connect(config.get('main', 'db'))

    try:
        filename = os.path.basename(args.path)
        with open(args.path) as source:
            conn.execute('insert into notes values (?, ?)',
                         (filename, source.read()))
        conn.commit()
        log.info('Successfully added %r', filename)
    except:
        log.exception('Something went wrong')

if __name__ == '__main__':
    main()

As you can see, the script above takes some command line arguments, loads some configuration from a file, sets up log handling and then loads a file into a database. While simple and effective, this script is hard to test. Even using the TempDirectory and Replacer helpers from the TestFixtures package, the only way to do so is to write one or more high level tests such as the following:

from testfixtures import TempDirectory, Replacer, OutputCapture
import sqlite3

class Tests(TestCase):

    def test_main(self):
        with TempDirectory() as d:
            # setup db
            db_path = d.getpath('sqlite.db')
            conn = sqlite3.connect(db_path)
            conn.execute('create table notes (filename varchar, text varchar)')
            conn.commit()
            # setup config
            config = d.write('config.ini', '''
[main]
db = %s
log = %s
''' % (db_path, d.getpath('script.log')), 'ascii')
            # setup file to read
            source = d.write('test.txt', 'some text', 'ascii')
            with Replacer() as r:
                r.replace('sys.argv', ['script.py', config, source, '--quiet'])
                main()
            # check results
            self.assertEqual(
                conn.execute('select * from notes').fetchall(),
                [('test.txt', 'some text')]
                )

The problem is that, in order to test the different paths through the small piece of logic at the end of the script, we have to work around all the set up and handling done by the rest of the script.

This also makes it hard to re-use parts of the script. It’s common for a project to have several scripts, all of which get some config from the same file, have the same logging options and often use the same database connection.

Encapsulating the re-usable parts of scripts

So, let’s start by looking at how these common sections of code can be extracted into re-usable functions that can be assembled by Mush into scripts:

from argparse import ArgumentParser, Namespace
from .configparser import RawConfigParser
from mush import Runner, requires, attr, item
import logging, os, sqlite3, sys

log = logging.getLogger()

def base_options(parser: ArgumentParser):
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('--quiet', action='store_true',
                        help='Log less to the console')
    parser.add_argument('--verbose', action='store_true',
                        help='Log more to the console')

def parse_args(parser: ArgumentParser):
    return parser.parse_args()

def parse_config(args: Namespace) -> 'config':
    config = RawConfigParser()
    config.read(args.config)
    return dict(config.items('main'))

def setup_logging(log_path, quiet=False, verbose=False):
    handler = logging.FileHandler(log_path)
    handler.setLevel(logging.DEBUG)
    log.addHandler(handler)
    if not quiet:
        handler = logging.StreamHandler(sys.stderr)
        handler.setLevel(logging.DEBUG if verbose else logging.INFO)
        log.addHandler(handler)

class DatabaseHandler:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
    def __enter__(self):
        return self
    def __exit__(self, type, obj, tb):
        if type:
            log.exception('Something went wrong')
            self.conn.rollback()

We start with a function that adds the options needed by all our scripts to an argparse parser. This uses the requires decorator to tell Mush that it must be called with an ArgumentParser instance. See Configuring Resources for more details.

Next, we have a function that calls parse_args() on the parser and returns the resulting Namespace. The parse_config() function reads configuration from a file specified as a command line argument and so requires the Namespace. Since the config is a dict, we configure it as a named rather than typed resource. See Named resources for more details.

Finally, there is a function that configures log handling and a context manager that provides a database connection and handles exceptions that occur by logging them and aborting the transaction. Context managers like this are handled by Mush in a specific way, see Context manager resources for more details.

Each of these components can be separately and thoroughly tested. The Mush decorations are inert to all but the Mush Runner, meaning that they can be used independently of Mush in whatever way is convenient. As an example, the following tests use a TempDirectory and a LogCapture to fully test the database-handling context manager:

class DatabaseHandlerTests(TestCase):

    def setUp(self):
        self.dir = TempDirectory()
        self.addCleanup(self.dir.cleanup)
        self.db_path = self.dir.getpath('test.db')
        self.conn = sqlite3.connect(self.db_path)
        self.conn.execute('create table notes '
                          '(filename varchar, text varchar)')
        self.conn.commit()
        self.log = LogCapture()
        self.addCleanup(self.log.uninstall)

    def test_normal(self):
        with DatabaseHandler(self.db_path) as handler:
            handler.conn.execute('insert into notes values (?, ?)',
                                 ('test.txt', 'a note'))
            handler.conn.commit()
        # check the row was inserted and committed
        curs = self.conn.cursor()
        curs.execute('select * from notes')
        self.assertEqual(curs.fetchall(), [('test.txt', 'a note')])
        # check there was no logging
        self.log.check()

    def test_exception(self):
        with ShouldRaise(Exception('foo')):
            with DatabaseHandler(self.db_path) as handler:
                handler.conn.execute('insert into notes values (?, ?)',
                                     ('test.txt', 'a note'))
                raise Exception('foo')
        # check the row not inserted and the transaction was rolled back
        curs = handler.conn.cursor()
        curs.execute('select * from notes')
        self.assertEqual(curs.fetchall(), [])
        # check the error was logged
        self.log.check(('root', 'ERROR', 'Something went wrong'))

Writing the specific parts of your script

Now that all the re-usable parts of the script have been abstracted, writing a specific script becomes a case of writing just two functions:

def args(parser):
    parser.add_argument('path', help='Path to the file to process')

def do(conn, path):
    filename = os.path.basename(path)
    with open(path) as source:
        conn.execute('insert into notes values (?, ?)',
                     (filename, source.read()))
    conn.commit()
    log.info('Successfully added %r', filename)

As you can imagine, this much smaller set of code is simpler to test and easier to maintain.

Assembling the components into a script

So, we now have a library of re-usable components and the specific callables we require for the current script. All we need to do now is assemble these parts into the final script. The full details of this are covered in Constructing runners but two common patterns are covered below.

Cloning

With this pattern, a “base runner” is created, usually in the same place that other re-usable parts of the original script are located:

from mush import Runner, requires, attr, item
base_runner = Runner(ArgumentParser)
base_runner.add(base_options, label='args')
base_runner.extend(parse_args, parse_config)
base_runner.add(setup_logging, requires(
    log_path = item('config', 'log'),
    quiet = attr(Namespace, 'quiet'),
    verbose = attr(Namespace, 'verbose')
))

The code above shows how to label a point, args in this case, enabling callables to be inserted at that point at a later time. See Labels for full details. It also shows some different ways of getting Mush to pass parts of an object returned from a previous callable to the parameters of another callable. See Using parts of a resource for full details.

Now, for each specific script, the base runner is cloned and the script-specific parts added to the clone leaving a callable that can be put in the usual block at the bottom of the script:

main = base_runner.clone()
main['args'].add(args, requires=ArgumentParser)
main.add(DatabaseHandler, requires=item('config', 'db'))
main.add(do,
         requires(attr(DatabaseHandler, 'conn'), attr(Namespace, 'path')))

if __name__ == '__main__':
    main()

Using a factory

This pattern is most useful when you have lots of scripts that all follow a similar pattern when it comes to assembling the runner from the common parts and the specific parts. For example, if all your scripts take a path to a config file and a path to a file that needs processing, you can write a factory function that returns a runner based on the callable that does the work as follows:

def options(parser):
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('--quiet', action='store_true',
                        help='Log less to the console')
    parser.add_argument('--verbose', action='store_true',
                        help='Log more to the console')
    parser.add_argument('path', help='Path to the file to process')

def make_runner(do):
    runner = Runner(ArgumentParser)
    runner.add(options, requires=ArgumentParser)
    runner.add(parse_args, requires=ArgumentParser)
    runner.add(parse_config, requires=Namespace)
    runner.add(setup_logging, requires(
        log_path = item('config', 'log'),
        quiet = attr(Namespace, 'quiet'),
        verbose = attr(Namespace, 'verbose')
    ))
    runner.add(DatabaseHandler, requires=item('config', 'db'))
    runner.add(
        do,
        requires(attr(DatabaseHandler, 'conn'), attr(Namespace, 'path'))
    )
    return runner

With this in place, the specific script becomes the do() function we abstracted above and a very short call to the factory:

main = make_runner(do)

A combination of the clone and factory patterns can also be used to get the best of both worlds. Setting up several base runners that are clones of a parent runner and having factories that take common callable patterns and return complete runners can be very powerful.

Testing

The examples above have shown how using Mush can make it easier to have smaller components that are easier to re-use and test, however care should still be taken with testing. In particular, it’s a good idea to have some integration tests that exercise the whole runner checking that it behaves as expected when all command line options are specified and when just the defaults are used.

When using the factory pattern, the factories themselves should be unit tested. It can also make tests easier to write by having a “testing runner” that sets up the required resources, such as database connections, while maybe doing some things differently such as not reading a configuration file from disk or using a LogCapture instead of file or stream log handlers.

Some specific tools that Mush provides to aid automated testing are covered in Testing.