Буду благодарен за помощь. Хочу данный код сделать в асинхронном стиле 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')```