Вот код для подсчета пользователей в боте.
database.py
from typing import Union
import asyncpg
from asyncpg import Connection
from asyncpg.pool import Pool
class Database:
def __init__(self):
self.pool: Union[Pool, None] = None
async def create(self):
self.pool = await asyncpg.create_pool(
user=config.DB_USER,
password=config.DB_PASS,
host=config.DB_HOST,
database=config.DB_NAME
)
async def execute(self, command, *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
async def create_table_users(self):
sql = """
CREATE TABLE IF NOT EXISTS Users (
id SERIAL PRIMARY KEY,
full_name VARCHAR(255) NOT NULL,
username varchar(255) NULL
);
"""
async def add_user_Users(self, full_name, username):
sql = '''INSERT INTO users (full_name, username)
VALUES($1, $2) returning *'''
return await self.execute(sql, full_name, username, fetchrow=True)
await self.execute(sql, execute=True)
async def select_all_users(self):
sql = "SELECT * FROM Users"
return await self.execute(sql, fetch=True)
async def select_all_users_id(self):
sql = "SELECT telegram_id FROM Users"
return await self.execute(sql, fetch=True)
async def select_user(self, **kwargs):
sql = "SELECT * FROM Users WHERE "
sql, parameters = self.format_args(sql, parameters=kwargs)
return await self.execute(sql, *parameters, fetchrow=True)
async def count_users(self):
sql = "SELECT COUNT(*) FROM Users"
return await self.execute(sql, fetchval=True)
config.py
DB_USER = env.str('DB_USER') # Имя пользователя db
DB_PASS = env.str('DB_PASS') # Пароль от db
DB_HOST = env.str('DB_HOST')
DB_NAME = env.str('DB_NAME') # Название db
start.py
@dp.message_handler(CommandStart())
async def register_user(message: types.Message):
try:
user = await db.add_user_Users(telegram_id=message.from_user.id,
full_name=message.from_user.full_name,
username=message.from_user.username)
except asyncpg.exceptions.UniqueViolationError:
user = await db.select_user(telegram_id=message.from_user.id)
count = await db.count_users()
# Забираем как список или как словарь
user_data = list(user)
user_data_dict = dict(user)
# Забираем напрямую как из списка или словаря
username = user.get("username")
full_name = user[2]
await message.answer(
"\n".join(
[
f'Привет, {message.from_user.full_name}!',
f'В базе <b>{count}</b> пользователей'
]))
PS Использую PostgreSQL, Pythyon 3.8+ (может и будет работать на 3.7, но не проверял), asyncpg.