aiopg simulate error

simulate.md

#!/usr/local/bin/python
__author__ = 'spouk'

from asyncio import coroutine
import asyncio
from aiohttp import web
import aiopg
from aiopg.sa import create_engine
import sqlalchemy as sa

#---------------------------------------------------------------------------
#   md
#---------------------------------------------------------------------------
metadata = sa.MetaData()

#---------------------------------------------------------------------------
#  model
#---------------------------------------------------------------------------

# теги связь постов с тэгами
simpler = sa.Table('simpler', metadata,
                 sa.Column('id', sa.Integer, ),
                 sa.Column('title', sa.VARCHAR(100), ),
                 )

dbconfig  = dict(
    user='anonym',
    database='tester',
    host='localhost',
    port=5432,
    password='anonym',
)



class Handlers(object):
    def __init__(self, server, db=None):
        self.db = db
        self.server = server

    @asyncio.coroutine
    def root(self, request):
        print(request)
        return web.Response(body="ROOT page".encode())


class Server(object):
    def __init__(self, dbconfig, host='127.0.0.1', port=8080):
        self.dbconfig = dbconfig
        self.port = port
        self.host = host
        self.server = None
        self.loop = None
        self.hand = Handlers(server=self)
        self.db = None
        # con
        self.count = 0
        self.db_exists = False

    @asyncio.coroutine
    def init(self, loop):
        app = web.Application(loop=loop, middlewares=[self.postgress_middleware()])
        app.router.add_route('GET', '/', self.hand.root)
        self.server = yield from loop.create_server(app.make_handler(),
                                            self.host, self.port)
        print("Server started at {}:{}".format(self.host, self.port))


    @asyncio.coroutine
    def update_database(self):
        with (yield from self.db) as conn:
            yield from conn.execute(simpler.insert().values(id=self.count, title="ssss"))

    @asyncio.coroutine
    def update_database_transit(self):
        with (yield from self.db) as conn:
            tr = yield from conn.begin()
            print("insert new record: ", self.count)
            dbs = simpler
            q = dbs.insert().values(
                id = self.count,
                title = "new liner {}".format(self.count)
            )
            yield from conn.execute(q)
            yield from tr.commit()

        self.count += 1


    def postgress_middleware(self):
        @coroutine
        def postgress_database_connector(app, handler):
            @coroutine
            def middleware(request):
                if not self.db_exists:
                    engine = yield from aiopg.sa.create_engine(**self.dbconfig)
                    self.db_exists = True
                    self.database = request.db = self.db = self.db = engine

                yield from self.update_database()
                yield from self.update_database_transit()

                response = yield from handler(request)
                return response
            return middleware
        return postgress_database_connector

    def run(self):
        self.loop = asyncio.get_event_loop()
        self.loop.run_until_complete(self.init(self.loop))
        try:
            self.loop.run_forever()
        except KeyboardInterrupt:
            pass

s = Server(dbconfig)
s.run()

./server.py

simulate load

wrk -d5s -t5 -c20 -v http://127.0.0.1:8080

get result and erros

<Request GET / >
<Request GET / >
<Request GET / >
insert new record:  1064
insert new record:  1064
Fatal error on aiopg connection: bad state in _ready callback
connection: <aiopg.connection.Connection object at 0x80831aeb8>
Fatal error on aiopg connection: bad state in _ready callback
connection: <aiopg.connection.Connection object at 0x808045668>