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)
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, first, last, attr, item
import logging, os, sqlite3, sys
log = logging.getLogger()
@requires(ArgumentParser)
def base_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')
@requires(last(ArgumentParser))
def parse_args(parser):
return parser.parse_args()
class Config(dict): pass
@requires(first(Namespace))
def parse_config(args):
config = RawConfigParser(dict_type=Config)
config.read(args.config)
return Config(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. It uses the last()
decorator to indicate that it should be the last callable to use the
ArgumentParser instance. See Resource usage periods
for more details.
The parse_config() function uses the first() decorator to
indicate that it needs first use of the command line arguments. It
uses those to read the specified configuration and return the
configuration as a dictionary. A sub-class of dict is used so
that other callables can easily indicate that they require the
configuration to be passed in.
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:
def setUp(self):
self.dir = TempDirectory()
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()
def tearDown(self):
self.log.uninstall()
self.dir.cleanup()
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 just writing 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, first, last, attr, item
base_runner = Runner(ArgumentParser, base_options, parse_args, parse_config)
base_runner.add(setup_logging,
log_path = item(first(Config), 'log'),
quiet = attr(first(Namespace), 'quiet'),
verbose = attr(first(Namespace), 'verbose'))
The code above 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.add(args, ArgumentParser)
main.add(DatabaseHandler, item(Config, 'db'))
main.add(do,
attr(DatabaseHandler, 'conn'),
attr(Namespace, 'path'))
if __name__ == '__main__':
main()
Using a factory¶
This pattern is most useful when you have several or more 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, ArgumentParser)
runner.add(parse_args, last(ArgumentParser))
runner.add(parse_config, first(Namespace))
runner.add(setup_logging,
log_path = item(first(Config), 'log'),
quiet = attr(first(Namespace), 'quiet'),
verbose = attr(first(Namespace), 'verbose'))
runner.add(DatabaseHandler, item(Config, 'db'))
runner.add(do,
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 and 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 intergration tests that excercise 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.