Making all the Boring Things Go Away

Chris Withers

Jump Trading

Who am I? Everyone has to write scripts, the tend to grow like mould...

A Simple Script for a Simple Task

import os, sqlite3, sys

conn = sqlite3.connect('mycoolapp.db')

filename = os.path.basename(sys.argv[1])
with open(sys.argv[1]) as source:
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()

but users want a test db

Another environment

import os, sqlite3, sys

if sys.argv[2] == 'prod':
    db = 'mycoolapp.db'
else:
    db = 'mycoolapp-test.db'

conn = sqlite3.connect(db)

filename = os.path.basename(sys.argv[1])
with open(sys.argv[1]) as source:
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()

but users can't tell what's going on

What's going on?

import os, sqlite3, sys

if sys.argv[2] == 'prod':
    db = 'mycoolapp.db'
else:
    db = 'mycoolapp-test.db'

print('connecting to:', db)
conn = sqlite3.connect(db)

filename = os.path.basename(sys.argv[1])
with open(sys.argv[1]) as source:
    print('opened {} to insert as {}', sys.argv[1], filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
print('yay!')

print statements everywhere

We need a real database!

from configparser import RawConfigParser
import os, psycopg2, sys

if sys.argv[2] == 'prod':
    db = 'mycoolapp.db'
else:
    db = 'mycoolapp-test.db'

config = RawConfigParser()
config.read(sys.argv[1])
user = config.get('db', 'user')
password = config.get('db', 'password')

print('connecting to {} as {}', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(sys.argv[3])
with open(sys.argv[3]) as source:
    print('opened {} to insert as {}', sys.argv[3], filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
print('yay!')

and that means we need a config file ...but our args are now crazy

I don't know how to use this!

from argparse import ArgumentParser
from configparser import RawConfigParser
import os, psycopg2, sys

parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
parser.add_argument('path', help='Path to the file to process')
args = parser.parse_args()

config = RawConfigParser()
config.read(args.config)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

print('connecting to {} as {}', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    print('opened {} to insert as {}', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
print('yay!')

add argparse

Its too verbose!

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)

"I only want that when I'm debugging" standard library logging - but briefly

I want to do something different!

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)

Copy and copy and copy...

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)
from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)
from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)
from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)
from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys


parser = ArgumentParser()
parser.add_argument('config', help='Path to .ini file')
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 = logging.getLogger()
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)

db = config.get('db', 'name')
user = config.get('db', 'user')
password = config.get('db', 'password')

log.debug('connecting to %s as %s', db, user)
conn = psycopg2.connect(dbname=db, user=user, password=password)

filename = os.path.basename(args.path)
with open(args.path) as source:
    log.debug('opened %s to insert as %s', args.path, filename)
    conn.execute('insert into notes values (?, ?)',
                 (filename, source.read()))
conn.commit()
log.info('successfully inserted %s', filename)

spot the bugs (quiet arg, no execute method on conn)

Uh Oh, We better test that!

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
@pytest.fixture(autouse=True)
def dir():
    with TempDirectory() as dir:
        yield dir

@pytest.fixture(autouse=True, scope='module')
def cursor():
    conn = psycopg2.connect(dbname=os.environ.get('DB_NAME'),
                            user=os.environ.get('DB_USER'),
                            password=os.environ.get('DB_PASSWORD'))
    cursor = conn.cursor()
    cursor.execute('drop table if exists notes')
    cursor.execute('create table notes (filename varchar, text varchar)')
    conn.commit()
    yield cursor

def test_script(dir, cursor):
    config_path = dir.write('config.ini', encoding='utf-8', data=dedent("""
    [main]
    log={log}

    [db]
    name={name}
    user={user}
    password={password}
    """.format(log=dir.getpath('log.txt'),
               name=os.environ.get('DB_NAME'),
               user=os.environ.get('DB_USER'),
               password=os.environ.get('DB_PASSWORD'))))

    file_path = dir.write('file.txt', encoding='utf-8', data='my note')

    with mock.patch('sys.argv', ['x', config_path, file_path]):
        with OutputCapture() as output:
            script()

    cursor.execute('select * from notes')
    compare(cursor.fetchall(), expected=[
        ('file.txt', 'my note')
    ])

    output.compare(
        'successfully inserted file.txt'
    )

Have to test at a high level. Getting coverage is a pain. Not checking log levels.

...and for each of the copies No code re-use, bugs are everywhere How did we get here? Where do we go next?

Anti-pattern 2: The Base Class

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
class BaseScript(object):

    def load_args(self):
        parser = ArgumentParser()
        parser.add_argument('config', help='Path to .ini file')
        parser.add_argument('-q', '--quiet', action='store_true')
        parser.add_argument('-v', '--verbose', action='store_true')
        self.add_args(parser)
        self.args = parser.parse_args()

    def add_args(self):
        pass

    def load_config(self):
        self.config = RawConfigParser()
        self.config.read(self.args.config)

    def setup_logging(self):
        handler = logging.FileHandler(self.config.get('main', 'log'))
        handler.setLevel(logging.DEBUG)
        self.log = logging.getLogger()
        self.log.addHandler(handler)
        self.log.setLevel(logging.DEBUG)
        if not self.args.quiet:
            handler = logging.StreamHandler(sys.stderr)
            handler.setLevel(logging.DEBUG if self.args.verbose else logging.INFO)
            self.log.addHandler(handler)

    _conn = None
    @property
    def conn(self):
        if self._conn is None:
            db = self.config.get('db', 'name')
            user = self.config.get('db', 'user')
            password = self.config.get('db', 'password')
            self.log.debug('connecting to %s as %s', db, user)
            conn = psycopg2.connect(dbname=db, user=user, password=password)
            self._conn = conn
        return self._conn

    def run(self):
        self.load_args()
        self.load_config()
        self.setup_logging()
        try:
            self.script(self)
            self.conn.commit()
        except:
            self.conn.rollback()
            self.log.exception('script failed')
        else:
            self.log.info('completed successfully')

    def script(self, parser):
        pass
class MyScript(BaseScript):

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

    def script(self):
        filename = os.path.basename(self.args.path)
        with open(self.args.path) as source:
            self.log.debug('opened %s to insert as %s', self.args.path, filename)
            cursor = self.conn.cursor()
            cursor.execute('insert into notes values (%s, %s)',
                           (filename, source.read()))
        self.log.info('successfully inserted %s', filename)


if __name__ == '__main__':
    MyScript().run()

What's so bad about that?

abstract into classes -> bells-and-whistles class end up with infinite plugin methods that are all called usually just test the 'do-it' method hard to test all the plugin points

Anti-pattern 3: Splitting Configuration

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    parser.add_argument('--db-name')
    parser.add_argument('--db-user')
    parser.add_argument('--db-pass')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = args.db_name or os.environ.get('DB_NAME') or config.get('db', 'name')
    user = args.db_user or os.environ.get('DB_USER') or config.get('db', 'user')
    password = args.db_pass or os.environ.get('DB_PASS') or config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()

splitting config across objects (config, args, env)

So how did we get here?

  • Multiple scripts ends up needing framework
  • Each script is usually a single function
  • Much shared config
import logging
import os

logger = logging.getLogger(__name__)

def insert_note(path, conn):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)

A tested function

import logging
import os

logger = logging.getLogger(__name__)

def insert_note(path, conn):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)
from testfixtures import TempDirectory, LogCapture, compare


@pytest.fixture(autouse=True)
def log():
    with LogCapture() as log:
        yield log

@pytest.fixture(autouse=True)
def dir():
    with TempDirectory() as dir:
        yield dir

@pytest.fixture(autouse=True, scope='module')
def conn():
    conn = psycopg2.connect(dbname=os.environ.get('DB_NAME'),
                            user=os.environ.get('DB_USER'),
                            password=os.environ.get('DB_PASSWORD'))
    cursor = conn.cursor()
    cursor.execute('drop table if exists notes')
    cursor.execute('create table notes (filename varchar, text varchar)')
    yield conn
    conn.rollback()


def test_insert_note(dir, conn, log):

    file_path = dir.write('file.txt', encoding='utf-8', data='my note')

    insert_note(file_path, conn)

    cursor = conn.cursor()
    cursor.execute('select * from notes')
    compare(cursor.fetchall(), expected=[
        ('file.txt', 'my note')
    ])

    log.check(
        ('code.insert_note', 'DEBUG',
         'opened {} to insert as file.txt'.format(file_path)),
        ('code.insert_note', 'INFO', 'successfully inserted file.txt')
    )

Components: Argument Parsing

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
from argparse import ArgumentParser, Namespace

def common_arguments(parser: ArgumentParser):
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    parser.add_argument('--db-name')
    parser.add_argument('--db-user')
    parser.add_argument('--db-password')

def parse_args(parser: ArgumentParser) -> Namespace:
    return parser.parse_args()

Components: Config

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
import yaml


def load_config(path):
    with open(path) as source:
        return yaml.safe_load(source)


def adjust_config(args: Namespace, config):
    for name in 'name', 'user', 'password':
        environ_value = os.environ.get('DB_'+name.upper())
        arg_value = getattr(args, 'db_'+name)
        value = arg_value or environ_value
        if value:
            config['db'][name] = value

user preferences woven in here mention configurator?

Components: Logging

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
import sys
from subprocess import list2cmdline

logger = logging.getLogger()

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

@contextmanager
def log_details():
    logger.info('Called as: %s', list2cmdline(sys.argv))
    try:
        yield
    except KeyboardInterrupt:
        logger.warning('keyboard interrupt')
        raise
    except SystemExit:
        logger.error('system exit')
        raise
    except:
        logger.exception('unhandled exception')

Components: Database

from argparse import ArgumentParser
from configparser import RawConfigParser
import logging, os, psycopg2, sys

def script():
    parser = ArgumentParser()
    parser.add_argument('config', help='Path to .ini file')
    parser.add_argument('path', help='Path to the file to process')
    parser.add_argument('-q', '--quiet', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    config = RawConfigParser()
    config.read(args.config)

    handler = logging.FileHandler(config.get('main', 'log'))
    handler.setLevel(logging.DEBUG)
    log = logging.getLogger()
    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)

    db = config.get('db', 'name')
    user = config.get('db', 'user')
    password = config.get('db', 'password')

    log.debug('connecting to %s as %s', db, user)
    conn = psycopg2.connect(dbname=db, user=user, password=password)

    filename = os.path.basename(args.path)
    with open(args.path) as source:
        log.debug('opened %s to insert as %s', args.path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    conn.commit()
    log.info('successfully inserted %s', filename)

if __name__ == '__main__':
    script()
logger = logging.getLogger(__name__)

@contextmanager
def handle_database(name, user, password):
    conn = psycopg2.connect(dbname=name, user=user, password=password)
    try:
        yield conn
        conn.commit()
    except:
        logger.exception('commit failed')
        conn.rollback()
        raise

How do we wire all that together?

Manually?

def script(func, add_args=None):

    parser = ArgumentParser()
    common_arguments(parser)
    if add_args:
        add_args(parser)
    args = parse_args(parser)

    config = load_config(args.config)
    adjust_config(args, config)

    setup_logging(config['log_path'], args.quiet, args.verbose)

    with log_details():

        with handle_database as conn:
            func(...)
import logging
import os

logger = logging.getLogger(__name__)

def insert_note(path, conn):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)

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

if __name__ == '__main__':
    script(insert_note, add_args)

what about those ellipsis? how do we know we got the stuff in the __main__ block right?

Mush

Runners

from mush import Runner

def func1():
    print('func1')

def func2():
    print('func2')


runner = Runner()
runner.add(func1)
runner.add(func2)
>>> runner()
func1
func2

Labels

from mush import Runner

def func1():
    print('func1')

def func2():
    print('func2')

def func2():
    print('func3')


runner = Runner()
runner.add(func1)
runner.add_label('middle')
runner.add(func2)

runner['middle'].add(func3)
>>> runner()
func1
func3
func2

Configuring Resources

def apple_tree():
    print('I made an apple')
    return Apple()

def magician(fruit: Apple) -> 'citrus':
    print('I turned {0} into an orange'.format(fruit))
    return Orange()

def juicer(fruit1: Apple, fruit2: 'citrus'):
    print('I made juice out of {0} and {1}'.format(fruit1, fruit2))
    return Juice()
>>> runner = Runner(apple_tree, magician, juicer)
>>> runner()
I made an apple
I turned an apple into an orange
I made juice out of an apple and an orange
a refreshing fruit beverage

Declarative Configuration

from mush import requires, returns

def apple_tree():
    print('I made an apple')
    return Apple()

@requires(Apple)
@returns('citrus')
def magician(fruit):
    print('I turned {0} into an orange'.format(fruit))
    return Orange()

@requires(fruit1=Apple, fruit2='citrus')
def juicer(fruit1, fruit2):
    print('I made juice out of {0} and {1}'.format(fruit1, fruit2))
    return Juice()
>>> runner = Runner(apple_tree, magician, juicer)
>>> runner()
I made an apple
I turned an apple into an orange
I made juice out of an apple and an orange
a refreshing fruit beverage

Default Configuration

def apple_tree() -> 'apple':
    print('I made an apple')
    return Apple()

def magician(apple) -> 'citrus':
    print('I turned {0} into an orange'.format(apple))
    return Orange()

def juicer(apple, citrus):
    print('I made juice out of {0} and {1}'.format(apple, citrus))
    return Juice()
>>> runner = Runner(apple_tree, magician, juicer)
>>> runner()
I made an apple
I turned an apple into an orange
I made juice out of an apple and an orange
a refreshing fruit beverage

Explicit Configuration

from mush import Runner, requires

def apple_tree():
    print('I made an apple')
    return Apple()

def magician(fruit):
    print('I turned {0} into an orange'.format(fruit))
    return Orange()

def juicer(fruit1, fruit2):
    print('I made juice out of {0} and {1}'.format(fruit1, fruit2))

runner = Runner()
runner.add(apple_tree)
runner.add(magician, requires=Apple, returns='citrus')
runner.add(juicer, requires(fruit1=Apple, fruit2='citrus'))
>>> runner()
I made an apple
I turned an apple into an orange
I made juice out of an apple and an orange

Back to our tested function...

import logging
import os

from argparse import ArgumentParser, Namespace
from mush import attr
from psycopg2.extensions import connection as Psycopg2Connection

logger = logging.getLogger(__name__)


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


def insert_note(path: attr(Namespace, 'path'), conn: Psycopg2Connection):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)

Assemble and Clone

def script(func, add_args=None):

    parser = ArgumentParser()
    common_arguments(parser)
    if add_args:
        add_args(parser)
    args = parse_args(parser)

    config = load_config(args.config)
    adjust_config(args, config)

    setup_logging(config['log_path'], args.quiet, args.verbose)

    with log_details():

        with handle_database as conn:
            func(...)
from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

script = Runner()
script.add(ArgumentParser)
script.add(common_arguments)

script.add_label('args')

script.add(parse_args)
script.add(load_config,
           requires=attr(Namespace, 'config'),
           returns='config')
script.add(adjust_config)

script.add_label('adjust_config')

script.add(setup_logging, requires(
    log_path=item('config', 'log_path'),
    quiet=attr(Namespace, 'quiet'),
    verbose=attr(Namespace, 'verbose'),
))
script.add(log_details)

script.add(handle_database, requires(
    name=item('config', 'db', 'name'),
    user=item('config', 'db', 'user'),
    password=item('config', 'db', 'password'),
))

script.add_label('body')

Assemble and Clone

from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

script = Runner()
script.add(ArgumentParser)
script.add(common_arguments)

script.add_label('args')

script.add(parse_args)
script.add(load_config,
           requires=attr(Namespace, 'config'),
           returns='config')
script.add(adjust_config)

script.add_label('adjust_config')

script.add(setup_logging, requires(
    log_path=item('config', 'log_path'),
    quiet=attr(Namespace, 'quiet'),
    verbose=attr(Namespace, 'verbose'),
))
script.add(log_details)

script.add(handle_database, requires(
    name=item('config', 'db', 'name'),
    user=item('config', 'db', 'user'),
    password=item('config', 'db', 'password'),
))

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


def insert_note(path: attr(Namespace, 'path'), conn: Psycopg2Connection):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)


main = script.clone()
main['args'].add(args)
main.add(insert_note)
if __name__ == '__main__':
    main()

Runner Factory

from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

script = Runner()
script.add(ArgumentParser)
script.add(common_arguments)

script.add_label('args')

script.add(parse_args)
script.add(load_config,
           requires=attr(Namespace, 'config'),
           returns='config')
script.add(adjust_config)

script.add_label('adjust_config')

script.add(setup_logging, requires(
    log_path=item('config', 'log_path'),
    quiet=attr(Namespace, 'quiet'),
    verbose=attr(Namespace, 'verbose'),
))
script.add(log_details)

script.add(handle_database, requires(
    name=item('config', 'db', 'name'),
    user=item('config', 'db', 'user'),
    password=item('config', 'db', 'password'),
))

script.add_label('body')
from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

def script(body, args=None):

    script = Runner()
    script.add(ArgumentParser)
    script.add(common_arguments)
    if args:
        script.add(args, requires=ArgumentParser)
    script.add(parse_args)

    script.add(load_config,
               requires=attr(Namespace, 'config'),
               returns='config')
    script.add(adjust_config)

    script.add(setup_logging, requires(
        log_path=item('config', 'log_path'),
        quiet=attr(Namespace, 'quiet'),
        verbose=attr(Namespace, 'verbose'),
    ))
    script.add(log_details)

    script.add(handle_database, requires(
        name=item('config', 'db', 'name'),
        user=item('config', 'db', 'user'),
        password=item('config', 'db', 'password'),
    ))

    script.add(body)

    return script

Runner Factory

from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

def script(body, args=None):

    script = Runner()
    script.add(ArgumentParser)
    script.add(common_arguments)
    if args:
        script.add(args, requires=ArgumentParser)
    script.add(parse_args)

    script.add(load_config,
               requires=attr(Namespace, 'config'),
               returns='config')
    script.add(adjust_config)

    script.add(setup_logging, requires(
        log_path=item('config', 'log_path'),
        quiet=attr(Namespace, 'quiet'),
        verbose=attr(Namespace, 'verbose'),
    ))
    script.add(log_details)

    script.add(handle_database, requires(
        name=item('config', 'db', 'name'),
        user=item('config', 'db', 'user'),
        password=item('config', 'db', 'password'),
    ))

    script.add(body)

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


def insert_note(path: attr(Namespace, 'path'), conn: Psycopg2Connection):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)


main = script(insert_note, args)

if __name__ == '__main__':
    main()

Testing the Base Runner

from argparse import ArgumentParser, Namespace
from mush import Runner, attr, item, requires

script = Runner()
script.add(ArgumentParser)
script.add(common_arguments)

script.add_label('args')

script.add(parse_args)
script.add(load_config,
           requires=attr(Namespace, 'config'),
           returns='config')
script.add(adjust_config)

script.add_label('adjust_config')

script.add(setup_logging, requires(
    log_path=item('config', 'log_path'),
    quiet=attr(Namespace, 'quiet'),
    verbose=attr(Namespace, 'verbose'),
))
script.add(log_details)

script.add(handle_database, requires(
    name=item('config', 'db', 'name'),
    user=item('config', 'db', 'user'),
    password=item('config', 'db', 'password'),
))

script.add_label('body')
  • This is still a pain to test
  • But you only have to do it once
  • Component tests do heavy lifting

Testing the Scripts

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


def insert_note(path: attr(Namespace, 'path'), conn: Psycopg2Connection):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)


main = script.clone()
main['args'].add(args)
main.add(insert_note)
if __name__ == '__main__':
    main()
  • How do we test these?
  • Have we wired up resources correctly?

Testing the Scripts

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


def insert_note(path: attr(Namespace, 'path'), conn: Psycopg2Connection):
    filename = os.path.basename(path)
    with open(path) as source:
        logger.debug('opened %s to insert as %s', path, filename)
        cursor = conn.cursor()
        cursor.execute('insert into notes values (%s, %s)',
                       (filename, source.read()))
    logger.info('successfully inserted %s', filename)


main = script.clone()
main['args'].add(args)
main.add(insert_note)
if __name__ == '__main__':
    main()
@pytest.fixture(autouse=True)
def log():
    with LogCapture(attributes=('levelname', 'getMessage')) as log:
        yield log

@pytest.fixture(autouse=True)
def dir():
    with TempDirectory() as dir:
        yield dir

@pytest.fixture(autouse=True, scope='module')
def conn():
    conn = psycopg2.connect(dbname=os.environ.get('DB_NAME'),
                            user=os.environ.get('DB_USER'),
                            password=os.environ.get('DB_PASSWORD'))
    cursor = conn.cursor()
    cursor.execute('drop table if exists notes')
    cursor.execute('create table notes (filename varchar, text varchar)')
    yield conn
    conn.rollback()
def run_with(main, conn, argv):
    runner = Runner(ArgumentParser)
    runner.extend(main.clone(added_using='args'))
    runner.add(lambda parser: parser.parse_args(argv),
               requires=ArgumentParser)
    runner.add(lambda: conn)
    runner.extend(main.clone(start_label='body'))
    runner()

def test_script(dir, conn, log):

    file_path = dir.write('file.txt', encoding='utf-8', data='my note')

    run_with(main, conn, argv=[file_path])

    cursor = conn.cursor()
    cursor.execute('select * from notes')
    compare(cursor.fetchall(), expected=[
        ('file.txt', 'my note')
    ])

    log.check(
        ('DEBUG', 'opened {} to insert as file.txt'.format(file_path)),
        ('INFO', 'successfully inserted file.txt')
    )

Summary

  • keep your tasks and framework separate
  • abstract and test framework components separately
  • have one source of framework configuration information
  • if using mush
    • use lots of labels
    • write a "run with" helper

Tools I've Used

mush

testfixtures

Options

command line processing

Options

config files

Options

validation

Questions

?

Thanks

Getting hold of me:

SpaceForward
Right, Down, Page DownNext slide
Left, Up, Page UpPrevious slide
POpen presenter console
HToggle this help