Как сделать единое подключение к бд (использую aiomysql)

Question

Стал использовать библиотеку aiomysql для асинхронного подключения к бд. Раньше использовал mysql.connector и подключался к бд только один раз. В документации aiomysql нет ни одного примера для подключения к бд всего один раз, там есть единый метод, в котором происходят и подключение и запросы. В своей программе я много работаю с бд и на взятие каких-либо данных из бд у меня несколько методов, не буду же я в каждом методе заново подключаться к базе данных. Как мне сделать подключение к бд одиножды и использовать его в других методах?

loop = asyncio.get_event_loop()

conn = await connect(host=host, port=port, user=user, password=password, db=db, 
                     charset='utf8', use_unicode=True, loop=loop)

async def add_user(user: usermodel):
    cur = await conn.cursor(prepared=True)

    try:
        await cur.execute('INSERT INTO Users (Phone, ...) VALUES (?, ...)',
                          (user.phone, ...))
    except aiomysql.Error as ex:
        print(f'Error insert: {ex}')
        sys.exit(1)

    await conn.commit()
    await cur.close()

Данный код демонстрирует то, что я хотел бы видеть на aiomysql

ps: формат подключения к бд в коде выше был реализован с помощью mariadb connector, такой же формат подключения использует и mysql connector, хочу такую же реализацию на aiomysql

Answer 1

settings.py:

class Settings:
    def __init__(self):
        self._conn = None

    @property
    def conn(self):
        return self._conn

    @conn.setter
    def conn(self, value):
        self._conn = value


settings = Settings()

main.py:

import ...
from settings import settings

loop = asyncio.get_event_loop()


async def main():
    if not settings.conn:
        try:
            settings.conn = await create_pool(host=HOST, port=PORT, user=USER, 
                                          password=PASSWORD,
                                          db=DATABASE, charset='utf8',
                                          use_unicode=True, loop=loop)
        except aiomysql.Error as e:
            print(f'Error connecting: {e}')
            sys.exit(1)



if __name__ == '__main__':
    loop.run_until_complete(main())

db.py:

import ...
from settings import settings


async def user_info(attr, val):
    data = None
    pool = settings.conn
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            try:
                await cur.executemany('SELECT phone, ... FROM Table' 
                                      f'WHERE {attr} = %s', (val,))
            except aiomysql.Error as ex:
                print(f'Error select: {ex}')
                sys.exit(1)

            data = await cur.fetchall()

    pool.close()
    await pool.wait_closed()
    return data

БЛОГ НА HUSL

Как сделать единое подключение к бд (использую aiomysql)

Ответы (1 шт):