Simple Python aiohttp Server with Configuration Parsing and Logging

example.conf

server_addr = "127.0.0.1"
server_port = 8888

# debug level, can be debug, error, info, ...
loglevel = "debug"

paths = {
    "route_set" : "/",
    "interface_get" : "/",
}

server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import asyncio
import aiohttp.web
import time
import sys
import os
import datetime
import argparse
import pprint
import subprocess
import logging

log = logging.getLogger()

# exit codes for shell, failure cones can be sub-devided
# if required and shell/user has benefit of this information
EXIT_OK      = 0
EXIT_FAILURE = 1

PROG_NAME = "TICNG"


def http_failure(ctx, msg):
    response_data = {'status': 'failure', "message": msg}
    body = json.dumps(response_data).encode('utf-8')
    return aiohttp.web.Response(body=body, content_type="application/json")


async def httpd_handler_route_set(request):
    ctx = request.app['ctx']
    try:
        request_data = await request.json()
    except json.decoder.JSONDecodeError:
        return http_failure("message": "data not properly formated")
    response_data = {'status': 'ok'}
    body = json.dumps(response_data).encode('utf-8')
    return aiohttp.web.Response(body=body, content_type="application/json")


def httpd_init(ctx, loop):
    conf = ctx['conf']
    app = aiohttp.web.Application(loop=loop)
    app['ctx'] = ctx
    path = conf['paths']['route_set']
    app.router.add_route('POST', path, httpd_handler_route_set)
    server = loop.create_server(app.make_handler(), conf['server_addr'], conf['server_port'])
    fmt = "HTTP IPC server started at http://{}:{}"
    log.error(fmt.format(conf['server_addr'], conf['server_port']))
    loop.run_until_complete(server)


def main(ctx):
    loop = asyncio.get_event_loop()
    httpd_init(ctx, loop)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        for task in asyncio.Task.all_tasks():
            task.cancel()
        loop.close()


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("-f", "--configuration", help="configuration", type=str, default=None)
    parser.add_argument("-v", "--verbose", help="verbose", action='store_true', default=False)
    args = parser.parse_args()
    if not args.configuration:
        emsg = "Configuration required, please specify a valid file path, exiting now\n"
        sys.stderr.write(emsg)
    return args


def load_configuration_file(args):
    config = dict()
    exec(open(args.configuration).read(), config)
    return config


def init_logging(conf):
    log_level_conf = "warning"
    if "loglevel" in conf:
        log_level_conf = conf['loglevel']
    numeric_level = getattr(logging, log_level_conf.upper(), None)
    if not isinstance(numeric_level, int):
        raise ConfigurationException('Invalid log level: {}'.format(numeric_level))
    logging.basicConfig(level=numeric_level, format='%(message)s')
    log.error("Logging level: {}".format(log_level_conf))


def conf_init():
    args = parse_args()
    conf = load_configuration_file(args)
    init_logging(conf)
    return conf, args


def ctx_init():
    return dict()


if __name__ == '__main__':
    sys.stderr.write("{}\n".format(PROG_NAME))
    conf, args = conf_init()
    ctx =  ctx_init()
    ctx['conf'] = conf
    ctx['args'] = args
    main(ctx)