Проблема не в том, что sigint не обрабатывается, а в том, что он поднимается в каждый форк и там все взрывается, поэтому в трейсе нарисована куча KeyboardInterrupt вместо нуля.
А если sigint в форках игнорировать, то похоже что все ништяк. По идее в кейсе с aiohttp то же самое должно прокатить.
initializer при запуске пула:app.process_pool = ProcessPoolExecutor(3, initializer=register_signal_handler)def register_signal_handler():
signal.signal(signal.SIGINT, lambda _, __: None)In [1]: s = [1,2,3]
In [2]: 1 <= (len(s)-1)
Out[2]: True
In [3]: 3 <= (len(s)-1)
Out[3]: FalseDBInterface ,который имеет стандартные методы добавить/удалить данные. Далее создаем класс SQLdb который реализует этот интерфейс.SQLdb и экземпляр класса Blog , передавая туда sqldb. Когда мы захотим переехать на монгу - нам не надо будет трогать класс блога. Мы всего лишь реализуем класс Mongodb для новой базы и будем опрокидывать его в блог вместо sqldb. Итог - блог не зависит от системы хранения вообще.from abc import ABC, abstractmethod
import random
class DBInterface(ABC):
@abstractmethod
def insert(self, data):
pass
@abstractmethod
def get(self, id):
pass
class SQLdb(DBInterface):
def insert(self, data):
print(f"inserted {data} from mysql")
return random.randint(0, 100)
def get(self, id):
print(f"get user {id} data from mysql")
return {'id': id, "data": "some_data"}
class Mongodb(DBInterface):
def insert(self, data):
print(f"inserted {data} from mongo")
return random.randint(0, 100)
def get(self, id):
print(f"get user {id} data from mongo")
return {'id': id, "data": "some_data"}
class Blog:
def __init__(self, db, blog_id):
self.db = db
self.blog_id = blog_id
def add_user(self, name):
self.db.insert(name)
def get_users(self):
return self.db.get(self.blog_id)
if __name__ == '__main__':
sql_db = SQLdb()
mongo_db = Mongodb()
blog_with_sql = Blog(db=sql_db, blog_id=1)
blog_with_sql.add_user("test") # Выведет inserted test from mysql
blog_with_mongo = Blog(db=mongo_db, blog_id=1)
blog_with_mongo.add_user("test") # Выведет inserted test from mongoSTATE_DICT )import config
import telebot
bot = telebot.TeleBot(config.token)
STATE_DICT = {}import config
import telebot
from telebot import types
bot = telebot.TeleBot(config.token)
STATE_DICT = {}
@bot.message_handler(commands=['start'])
def start(message):
markup = types.ReplyKeyboardMarkup(
one_time_keyboard=False,
resize_keyboard=True
)
markup.add('Кнопка1')
markup.add('Кнопка2')
question = 'Выбери'
chat_id = message.chat.id
STATE_DICT[chat_id] = 'START'
bot.send_message(chat_id, question, reply_markup=markup)
def start_work(message):
chat_id = message.chat.id
markup = types.ReplyKeyboardMarkup(
one_time_keyboard=False,
resize_keyboard=True
)
if message.text == 'Кнопка1':
STATE_DICT[chat_id] = 'VAR1'
markup.add('Кнопка3')
markup.add('Кнопка4')
bot.send_message(chat_id, 'Вы выбрали ответ 1', reply_markup=markup)
else:
STATE_DICT[chat_id] = 'VAR2'
markup.add('Кнопка5')
markup.add('Кнопка6')
bot.send_message(chat_id, 'Вы выбрали ответ 2', reply_markup=markup)
def var1(message):
chat_id = message.chat.id
if message.text == 'Кнопка3':
bot.send_message(chat_id, 'Конец с кнопкой3')
else:
STATE_DICT[chat_id] = 'VAR2'
bot.send_message(chat_id, 'Конец с кнопкой2')
def var2(message):
chat_id = message.chat.id
if message.text == 'Кнопка5':
bot.send_message(chat_id, 'Конец с кнопкой5')
else:
STATE_DICT[chat_id] = 'VAR2'
bot.send_message(chat_id, 'Конец с кнопкой6')
@bot.message_handler(func=lambda message: True)
def start_finite_machine(message):
states = {
'START': start_work,
'VAR1': var1,
'VAR2': var2,
}
chat_id = message.chat.id
states[STATE_DICT[chat_id]](message)
if __name__ == '__main__':
bot.polling(none_stop=True)@bot.message_handler(func=lambda message: True)
def start_finite_machine(message):
if STATE_DICT[chat_id] == 'position_1':
position_1(message)
elif STATE_DICT[chat_id] == 'position_2':
position_2(message)
elif STATE_DICT[chat_id] == 'position_3':
position_3(message)del msg и вроде перестало.
from .stats.views import sorting_task_response - так тоже выбивает ошибку?