Асинхронный модуль работы с базой данных postgresql на python

Question

Буду благодарен за помощь. Хочу данный код сделать в асинхронном стиле c использованием asyncpg? В данном примере https://github.com/alexey-goloburdin/telegram-finance-bot модуль работы с базой данных представлен в синхронном стиле, однако используется асинхронная библиотека aiogram. Имеет ли смысл переписывать мне мой модуль? Будет ли давать это прирост производительности? Сейчас пока даже не понимаю с чего подступится. В частности нет понимания как должен выглядеть декоратор command, методы: _get_connection, select_table, init_db, insert_db_exhibition_info.

import psycopg2

def command(func):
    def wrapper(self, *args, **kwargs):
        cursor = self._get_connection().cursor()            # открываем курсор
        res = func(self, *args, cursor=cursor, **kwargs)    # выполняем функцию
        self._get_connection().commit()                     # коммит, если он нужен
        cursor.close()                                      # закрываем курсор
        return res
    return wrapper


class DbHelper():
    def __init__(self, db_name: str, user: str, password: str):
        self.db_name = db_name
        self.user = user
        self.password = password
        self._connection = None    # connection object


    def _get_connection(self):
        if not self._connection:   # ленивая инициализация соединения
            self._connection = psycopg2.connect(database=self.db_name, user=self.user, password=self.password)
        return self._connection

    def __del__(self):
        if self._connection:        # закрытие соединения при удалении объекта как пример безопасной работы
            self._connection.close()

    @command                        # создаем новые таблицы в БД
    def init_db(self, cursor, force: bool = False):
        if force:
            cursor.execute('DROP TABLE IF EXISTS dogs_info')

            self._connection.commit()

        # таблица с данными о собаках
        cursor.execute('''CREATE TABLE IF NOT EXISTS dogs_info (
                           id              SERIAL PRIMARY KEY,
                           dogs_name       VARCHAR(30),
                           dogs_age        INTEGER NOT NULL,
                           prize_place     INTEGER''')

    @command
    def insert_db_exhibition_info(self, cursor, dogs_name, dogs_age, prize_place):            # вносим в таблицу инофрмацию о собаках
        cursor.execute('INSERT INTO dogs_info (dogs_name, dogs_age, prize_place) VALUES (%s, %s, %s)',
                       (dogs_name, dogs_age, prize_place))

    @command
    def select_table(self, cursor):
        cursor.execute('SELECT * FROM dogs_info')
        return cursor.fetchall()

    @command
    def delete_table_dogs_info(self, cursor):
        cursor.execute('DELETE FROM dogs_info')```

Answer 1

Имеет ли смысл переписывать мне мой модуль?

Это необходимо сделать. Здесь я многим повторю свой ответ здесь. При использовании asyncio любые вызовы синхронных IO-bound операций в асинхронных (корутинах/сопрограммах) ни к чему хорошему не приводят и сводят на нет целесообразность использования asyncio. Главная причина: все у нас выполняется в одном потоке.

Допустим, есть такой псевдокод.

@message_handler
async def wellcome():
    get_data_from_db()
    await reply_to_user()

На каждое входящее сообщение, будет вызываться этот обработчик. А так как он объявлен с использованием async - это корутина.

Здесь очевидно, что get_data_from_db реализована синхронно, потому что вызывается без await. Ну и это точно IO-bound операция, так как мы будем ждать завершения операций ввода/вывода после обращения к базе данных. Допустим, что пришло 10 сообщений - значит у нас создается и запускается 10 корутин.

Как они будут выполняться?

Под капотом у нас есть event-loop, который следит за выполнением корутин. Он выстраивает их в очередь и запускает. Работает всегда только одна, остальные в очереди.

Сначала первая сделает запрос в базу и... Будет ждать. Остальные 9 рады бы поделать какую-нибудь полезную работу, например, приступить к подготовке запроса и его отправке. Однако у нас нет параллельных потоков, корутины исполняются последовательно, но event-loop реализует механизм переключения между разными корутинами, чтобы выполнять все это совместно.

Если корутина заблокировала поток управления, то event-loop никогда не узнает, что вот сейчас мы не готовы сразу продолжить работу и можно пока другую корутину из очереди позвать поработать. Корутина должна сама сказать. Как?

С помощью await, однако произвольную фунцию нельзя вызвать с await. Это должна быть корутина. Следовательно get_data_from_db надо объявить с использованием async и в ней использовать только неблокирующие операции, т.е. использовать специальные библиотеки с поддержкой асинхронности, коей является asyncpg.

Будет ли давать это прирост производительности?

Будет, за счет конкурентности. Процессор будет выполнять больше полезной работы, а не ждать ввода вывода. Это обеспечит гораздо большую пропускную способность, ибо клиенты ждущие данных из базы не будут блокировать тех, для которых уже все готово.

PS: Код из вашего примера как раз вызывает синхронные IO-bound перации в корутинах-обработчиках. Использовать встроенный sqlite3 совершенно не уместно. Тут нужно было что-то вроде aiosqlite.

PPS: До изобретения asyncio использовали другую библиотеку (pyTelegramBotAPI), которая не задействует асинхронность, а обработка каждого входящего сообщения отдается на исполнение пулом потоков. При использовании IO-bound операций мы не испытываем проблем с GIL.

PPPS: Асинхронная программа должна быть асинхронная везде. Любой синхронный вызов надо уносить в отдельный поток и запускать с использованием метода run_in_executor, чтобы не сломать основную идею переключения между корутинами.

По поводу того, как должен выглядеть декоратор.

Вспомним, что запись:

@decorate
def func():
    pass

Эквивалентна:

def func():
    pass

func = decorate(func)

Чтобы обернуть корутину, надо из декоратора вернуть корутину

def decorate(func):
    async def wrapper(*args, **kwargs):
        # Получаем соединение
        res = await func(*args, **kwargs)
        # Коммитим
        return res
    return wrapper

И тогда можно будет нормально задекорировать любую асинхронную функцию.

А методам DBHelper достаточно добавить async к def и дальше руководствоваться API библиотеки asyncpg. Хорошо, что оно почти не отличается от psycopg, однако отличия есть и это может породить довольно долгую кропотливую отладку.

Там, например, cursor по-другому используется. К объекту подключения к базе можно сразу execute делать, для транзакций там надо использовать контекстный менеджер async with и все такое. Тут лучше самому во всем разобраться и накопить конкретных вопросов.

Но ни в коем случае нельзя __init__ и __del__ объявлять async. Это служебные методы, которые вызываются не нами, они по своему строению синхронные. __init__ можно оставить как есть, а закрытие соединения делать явно, отдельным методом, по крайней мере, пока, глядя на получающийся код, не придут в голову варианты получше.

БЛОГ НА HUSL

Асинхронный модуль работы с базой данных postgresql на python

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