Right way to connect to mysql use connection pool in async falcon app

[icebearrrr]

In sync falcon app, I think i could write a something like pool.py to create a mysql pool for all other modules to import from:

def create_pool():
    pool = create_mysql_pool
    return pool
pool = create_pool()

Then other modules could get connection from pool instead of connecting to db and close it for every request.
But in async app, i'm a little confused. I want to know the right way to do it.
What i can do now is to create mysql connection in middleware and set it as an attribute like context to falcon.asgi.Request object in process_request, and then close it in process_response:

class AsyncMysqlMiddleware:
    async def process_request(self, req, resp):
        req.context.conn = await aiomysql.connect(**db)
    async def process_response(self, req, resp, *args):
        await req.context.conn.close()

[vytas7]

Hi @icebearrrr!
I've never used aiomysql, so take this with a grain of salt, as I don't know what the "right" way to work with it is.

With the above disclaimer in mind, in ASGI, you can make use of startup and shutdown events, that are easy to implement with Falcon middleware. You can put async initialization code into a process_startup(...) method, as also showcased in our ASGI tutorial.

So we could create an aiomysql pool, and then either use process_request and process_response to automatically draw connections from pool and store them in req.context, or use the pool directly in a responder. The following works for me with both approaches:

import logging

import aiomysql
import falcon.asgi


# NOTE: Useful since ASGI doesn't otherwise have anything like wsgierrors
logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s', level=logging.INFO)


class AsyncMysqlMiddleware:
    def __init__(self):
        self._pool = None

    async def process_startup(self, scope, event):
        self._pool = await aiomysql.create_pool(
            host='localhost', port=3306, user='...', password='...',
            db='test')

    async def process_request(self, req, resp):
        req.context.pool = self._pool

        try:
            req.context.conn = await self._pool.acquire()
        except Exception:
            req.context.conn = None
            raise

    async def process_response(self, req, resp, resource, req_succeeded):
        if req.context.conn:
            self._pool.release(req.context.conn)


class Items:
    async def on_get(self, req, resp):
        cur = await req.context.conn.cursor()
        await cur.execute('SELECT value FROM numbers')
        rows = await cur.fetchall()
        await cur.close()

        resp.media = [row[0] for row in rows]

    async def on_get_with_pool(self, req, resp):
        async with req.context.pool.acquire() as conn:
            cur = await conn.cursor()
            await cur.execute('SELECT value FROM numbers')
            rows = await cur.fetchall()
            await cur.close()

        resp.media = [row[0] for row in rows]


app = falcon.asgi.App(middleware=[AsyncMysqlMiddleware()])
app.add_route('/items', Items())
app.add_route('/items2', Items(), suffix='with_pool')

[CaselIT]

@vytas7 should the answer be adapted to a faq item?

[vytas7]

Sure! Although this one can probably wait for the outcome of #1938, because technically aioredis 2 has removed await from the pool initializer. But I guess we can invent a use case for process_startup anyway 🙂

[CaselIT]

we could use a hypothetical library similar to aiomysql

[vytas7]

In the FAQ answer -- yeah, but it's also useful to link to the ASGI tutorial to show how it looks like in practice.

[CaselIT]

here you go #1982

[vytas7]

Thanks to @CaselIT, this has now been added to our FAQ: How do I manage my database connections with ASGI? w00t!