Simplest web service that can kill pgbouncer

web_service.py

import asyncio
import argparse
import functools
import uuid

import aiohttp.web
import asyncpg
import uvloop



class App(aiohttp.web.Application):
    def __init__(self, args):
        super().__init__()
        self.pg_host = args.pg_host
        self.pg_port = args.pg_port
        self.pg_db = args.pg_db
        self.pg_user = args.pg_user
        self.pg_password = args.pg_password

        self.on_startup.append(self._create_pool)

    async def _create_pool(self, app):
        self.pool = await asyncpg.create_pool(
            host=self.pg_host,
            port=self.pg_port,
            database=self.pg_db,
            user=self.pg_user,
            password=self.pg_password,
        )


async def handler(request):
    print('handler start')
    pool = request.app.pool  # type: asyncpg.pool.Pool
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.fetch('SELECT pg_sleep(1);')
    print('handler end')
    return aiohttp.web.Response()


def parse_arguments(argv=None):
    parser = argparse.ArgumentParser()

    parser.add_argument('--host', default='0.0.0.0')
    parser.add_argument('--port', type=int, default=8080)
    parser.add_argument('--path')

    parser.add_argument('--pg-host', default='localhost')
    parser.add_argument('--pg-port', type=int, default=6432)
    parser.add_argument('--pg-db', required=True)
    parser.add_argument('--pg-user', required=True)
    parser.add_argument('--pg-password', required=True)

    return parser.parse_args(argv)


def main():
    args = parse_arguments()
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)

    app = App(args)
    app.router.add_get('/', handler)
    app.freeze()

    aiohttp.web.run_app(app, host=args.host, port=args.port, path=args.path)


if __name__ == '__main__':
    main()