 |
13.01.2022, 18:46
|
|
Участник форума
Регистрация: 09.01.2020
Сообщений: 134
С нами:
3340178
Репутация:
33
|
|
Всех приветствую, многие используют разные реализации для работы с БД, я же использую все в одном файле, и всегда использую его в каждом своем проекте, сейчас же хочу показать данный простой способ реализации вам.
Для начала создадим отдельный файлик в котором будем хранить свой класс для работы с БД, у меня это 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
(
)
Надеюсь для кого-то это будет полезным. |
|
|
28.01.2022, 21:10
|
|
Участник форума
Регистрация: 13.02.2019
Сообщений: 183
С нами:
3815624
Репутация:
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
(
)
Надеюсь для кого-то это будет полезным.
Паста со слитых 3483493482384 раз курсов никому не будет полезна
|
|
|
|
|
|
Предыдущая тема
Следующая тема
|
Здесь присутствуют: 1 (пользователей: 0 , гостей: 1)
|
|
|
|
Линейный вид
