Использование asyncpg пула соединений с Sanic

Question

Я почистил несколько примеров в Интернете и придумал это. Этот код работает, но мне интересно, есть ли лучший способ сделать это?

user.py:


from asyncpg import create_pool
from sanic import Blueprint

bp = Blueprint('dp')

class pg:
    def __init__(self, pg_pool):
        self.pg_pool = pg_pool

    async def fetch(self, sql, *args, **kwargs):
        async with self.pg_pool.acquire() as connection:
            return await connection.fetch(sql, *args, **kwargs)

    async def execute(self, sql, *args, **kwargs):
        async with self.pg_pool.acquire() as connection:
            return await connection.execute(sql, *args, **kwargs)

@bp.listener('before_server_start')
async def init_pg(app, loop):
    """
    Init Postgresql DB.
    """
    bp.pg_pool = await create_pool(
        **app.config.PG_CFG,
        max_inactive_connection_lifetime=60,
        min_size=1,
        max_size=3,
        loop=loop,
    )
    bp.pg = pg(bp.pg_pool)
    print('-------- setup connection pool --------')

Используйте класс pg сейчас в webapp.py

webapp.py:


@app.route("/")
async def root(req):
    result = await app.pg.fetch('SELECT * FROM foo')

Answer 1

Я не делал это сам, используя postgres, однако я делал с Redis нечто подобное. Кроме того, я знаю, что другие люди использовали аналогичные установки, используя before_server_start создать бассейн.

Другим вариантом будет

Взгляните на: https://github.com/Skyscanner/aiotask-context и https://github.com/MagicStack/contextvars. Возможно, они могут предложить помощь в присоединении чего-то вроде подключения к петле.