Answer the question
In order to leave comments, you need to log in
How to call the id of each user aiogram, sqlite?
How to complete this handler? I don't know how to call each user's id in turn to send to all authorized users from the sqlite database. All functions from a DB and handeler'a code resulted. Pycharm writes: UNIQUE constraint failed: Users.id
@dp.message_handler(user_id=admin_id, commands=['tellall'])
async def mailing(message: types.Message):
await message.answer(text="Пришлите текст рассылки")
await Mailing.Text.set()
@dp.callback_query_handler(user_id=admin_id, state=Mailing.Text)
async def mailing_start(call: types.CallbackQuery, state: FSMContext):
text = message.text
await state.update_data(text=text)
await state.reset_state()
users = db.select_all_users()
for user in users:
try:
await bot.send_message(chat_id=db.select_user(),
text=text)
await sleep(0.3)
except Exception:
pass
await call.message.answer("Рассылка выполнена.")
class Database:
def __init__(self, path_to_db="main.db"):
self.path_to_db = path_to_db
@property
def connection(self):
return sqlite3.connect(self.path_to_db)
def execute(self, sql: str, parameters: tuple = None, fetchone=False, fetchall = False, commit = False):
if not parameters:
parameters = tuple()
connection = self.connection
connection.set_trace_callback(logger)
cursor = connection.cursor()
data = None
cursor.execute(sql, parameters)
if commit:
connection.commit()
if fetchone:
data = cursor.fetchone()
if fetchall:
data = cursor.fetchone()
connection.close()
return data
def create_table_users(self):
sql = """
CREATE TABLE Users (
id int NOT NULL,
Name varchar(255) NOT NULL,
email varchar(255),
PRIMARY KEY (id)
);
"""
self.execute(sql, commit=True)
def add_user(self, id: int, name:str, email: str = None):
sql = 'INSERT INTO Users(id,Name, email) VALUES(?, ?, ?)'
parameters = (id, name, email)
self.execute(sql, parameters=parameters, commit=True)
def select_all_users(self):
sql = 'SELECT * FROM Users'
return self.execute(sql, fetchall=True)
@staticmethod
def format_args(sql, parameters: dict):
sql += " AND ".join([
f'{item} = ?' for item in parameters
])
return sql, tuple(parameters.values())
def select_user(self, **kwargs):
sql = 'SELECT * FROM Users WHERE '
sql, parameters = self.format_args(sql,kwargs)
return self.execute(sql,parameters,fetchall=True)
def count_users(self):
return self.execute("SELECT COUNT(*) FROM Users;", fetchone= True)
def update_email(self, email,id):
sql = "UPDATE Users SET email=? WHERE id=?"
return self.execute(sql, parameters=(email, id), commit=True)
def delete_users(self):
self.execute("DELETE FROM Users WHERE True")
Answer the question
In order to leave comments, you need to log in
1. Write a full traceback of the error so that we don't have to guess.
2. And most importantly, does it bother you that you have async, but do you work with the database in synchronous mode?
ZY also except Exception: with pass, bothered.
You can use the owl carousel, there is such an implementation ... or you can make it yourself by linking the id with the element's for
Didn't find what you were looking for?
Ask your questionAsk a Question
731 491 924 answers to any question