Всех приветствую, многие используют разные реализации для работы с БД, я же использую все в одном файле, и всегда использую его в каждом своем проекте, сейчас же хочу показать данный простой способ реализации вам.
Для начала создадим отдельный файлик в котором будем хранить свой класс для работы с БД, у меня это postgres.py
Импортируем необходимые библиотеки:
После создадим сам класс и добавим в него метод инициализации:
Тут мы добавляем переменную pool, чтобы хранить в ней данные о нашем подключении.
Далее добавим само подключение к БД через asyncpg:
Тут мы нашу переменную pool добавляем пул подключения к БД Постгри.
И теперь самая главная функция, которая будет ответственна за запросы к БД Постгри:
Вызывая execute мы отправляем ему необходимый для нас запрос и выставляем аргумент который для нас необходим, пример:
В конец после создания класса добавляем:
Далее отправляемся к нашему main(if __name__ == '__main__':)
Будьте внимательны, повторный вызов get_event_loop может вызвать ошибку.
Полный код класса:
Надеюсь для кого-то это будет полезным.
Для начала создадим отдельный файлик в котором будем хранить свой класс для работы с БД, у меня это 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('данные для подключения')И теперь самая главная функция, которая будет ответственна за запросы к БД Постгри:
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()Надеюсь для кого-то это будет полезным.
