Гайд Асинхронный PostgresSQL или простое решение для работы с БД

Толкин

Активный
Автор темы
133
83
Всех приветствую, многие используют разные реализации для работы с БД, я же использую все в одном файле, и всегда использую его в каждом своем проекте, сейчас же хочу показать данный простой способ реализации вам.

Для начала создадим отдельный файлик в котором будем хранить свой класс для работы с БД, у меня это postgres.py

Импортируем необходимые библиотеки:
Python:
from typing import Union

import asyncpg
from asyncpg import Connection
from asyncpg.pool import Pool

После создадим сам класс и добавим в него метод инициализации:
Python:
class DataBaseClass:
    def __init__(self):
        self.pool: Union[Pool, None] = None

Тут мы добавляем переменную pool, чтобы хранить в ней данные о нашем подключении.
Далее добавим само подключение к БД через asyncpg:
Python:
    async def create_pool(self):
        self.pool = await asyncpg.create_pool('данные для подключения')
Тут мы нашу переменную pool добавляем пул подключения к БД Постгри.

И теперь самая главная функция, которая будет ответственна за запросы к БД Постгри:
Python:
    async def execute(self, command: str, *args,
                      fetch: bool = False,
                      fetchval: bool = False,
                      fetchrow: bool = False,
                      execute: bool = False):
        async with self.pool.acquire() as connection:
            connection: Connection
            async with connection.transaction():
                if fetch:
                    result = await connection.fetch(command, *args)
                elif fetchval:
                    result = await connection.fetchval(command, *args)
                elif fetchrow:
                    result = await connection.fetchrow(command, *args)
                elif execute:
                    result = await connection.execute(command, *args)
        return result

Вызывая execute мы отправляем ему необходимый для нас запрос и выставляем аргумент который для нас необходим, пример:
Python:
result = await DataBase.execute("SELECT username, balance FROM users WHERE chat_id = $1",
                                      message.from_user.id, fetchval=True)
print('Ваш баланс:', result[0]['balance'])
print('Ваш ник:', result[0]['username'])

Python:
result = await DataBase.execute("SELECT id, username FROM users ",
                                      fetchval=True)

for row in result:
    print(f'ID: {row["id"]} | Имя: {row["username"]}')

В конец после создания класса добавляем:
Python:
DataBase = DataBaseClass()

Далее отправляемся к нашему main(if __name__ == '__main__':)
Python:
loop = asyncio.get_event_loop()
loop.run_until_complete(DataBase.create_pool())

Будьте внимательны, повторный вызов get_event_loop может вызвать ошибку.

Полный код класса:
Python:
from typing import Union

import asyncpg
from asyncpg import Connection
from asyncpg.pool import Pool


class DataBaseClass:
    def __init__(self):
        self.pool: Union[Pool, None] = None

    async def create_pool(self):
        self.pool = await asyncpg.create_pool('Ваше подключение')

    async def execute(self, command: str, *args,
                      fetch: bool = False,
                      fetchval: bool = False,
                      fetchrow: bool = False,
                      execute: bool = False):
        async with self.pool.acquire() as connection:
            connection: Connection
            async with connection.transaction():
                if fetch:
                    result = await connection.fetch(command, *args)
                elif fetchval:
                    result = await connection.fetchval(command, *args)
                elif fetchrow:
                    result = await connection.fetchrow(command, *args)
                elif execute:
                    result = await connection.execute(command, *args)
        return result


DataBase = DataBaseClass()

Надеюсь для кого-то это будет полезным.
 
  • Злость
Реакции: F0RQU1N