Handlers and Middlewares processing union

Call for handlers now union in a single function for future extension.

Plus minor fixes in storages.
This commit is contained in:
Badiboy 2022-09-17 11:57:12 +03:00
parent 8d9dfcfac8
commit da639dd1f6
6 changed files with 106 additions and 93 deletions

View File

@ -469,6 +469,7 @@ class TeleBot:
webhook_url = "{}://{}:{}/{}".format(protocol, listen, port, url_path) webhook_url = "{}://{}:{}/{}".format(protocol, listen, port, url_path)
if certificate and certificate_key: if certificate and certificate_key:
# noinspection PyTypeChecker
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(certificate, certificate_key) ssl_ctx.load_cert_chain(certificate, certificate_key)
@ -5875,7 +5876,7 @@ class TeleBot:
return False return False
# middleware check-up method # middleware check-up method
def _check_middleware(self, update_type): def _check_middlewares(self, update_type):
""" """
Check middleware Check middleware
@ -5889,100 +5890,111 @@ class TeleBot:
def _run_middlewares_and_handler(self, message, handlers, middlewares, update_type): def _run_middlewares_and_handler(self, message, handlers, middlewares, update_type):
""" """
This class is made to run handler and middleware in queue. This class is made to run handlers and middlewares in queue.
:param handler: handler that should be executed. :param message: received message (update part) to process with handlers and/or middlewares
:param middleware: middleware that should be executed. :param handlers: all created handlers (not filtered)
:param middlewares: middlewares that should be executed (already filtered)
:param update_type: handler/update type (Update field name)
:return: :return:
""" """
data = {}
params =[]
handler_error = None
skip_handlers = False
if middlewares: if not self.use_class_middlewares:
for middleware in middlewares: for message_handler in handlers:
if middleware.update_sensitive: if self._test_message_handler(message_handler, message):
if hasattr(middleware, f'pre_process_{update_type}'): self._exec_task(message_handler['function'], message, pass_bot=message_handler['pass_bot'], task_type='handler')
result = getattr(middleware, f'pre_process_{update_type}')(message, data)
else:
logger.error('Middleware {} does not have pre_process_{} method. pre_process function execution was skipped.'.format(middleware.__class__.__name__, update_type))
result = None
else:
result = middleware.pre_process(message, data)
# We will break this loop if CancelUpdate is returned
# Also, we will not run other middlewares
if isinstance(result, CancelUpdate):
return
elif isinstance(result, SkipHandler):
skip_handlers = True
if handlers and not(skip_handlers):
try:
for handler in handlers:
process_handler = self._test_message_handler(handler, message)
if not process_handler: continue
for i in inspect.signature(handler['function']).parameters:
params.append(i)
if len(params) == 1:
handler['function'](message)
elif "data" in params:
if len(params) == 2:
handler['function'](message, data)
elif len(params) == 3:
handler['function'](message, data=data, bot=self)
else:
logger.error("It is not allowed to pass data and values inside data to the handler. Check your handler: {}".format(handler['function']))
return
else:
data_copy = data.copy()
for key in list(data_copy):
# remove data from data_copy if handler does not accept it
if key not in params:
del data_copy[key]
if handler.get('pass_bot'):
data_copy["bot"] = self
if len(data_copy) > len(params) - 1: # remove the message parameter
logger.error("You are passing more parameters than the handler needs. Check your handler: {}".format(handler['function']))
return
handler["function"](message, **data_copy)
break break
except Exception as e: else:
handler_error = e data = {}
if self.exception_handler: params =[]
self.exception_handler.handle(e) handler_error = None
else: skip_handlers = False
logging.error(str(e))
logger.debug("Exception traceback:\n%s", traceback.format_exc())
if middlewares: if middlewares:
for middleware in middlewares: for middleware in middlewares:
if middleware.update_sensitive: if middleware.update_sensitive:
if hasattr(middleware, f'post_process_{update_type}'): if hasattr(middleware, f'pre_process_{update_type}'):
getattr(middleware, f'post_process_{update_type}')(message, data, handler_error) result = getattr(middleware, f'pre_process_{update_type}')(message, data)
else:
logger.error('Middleware {} does not have pre_process_{} method. pre_process function execution was skipped.'.format(middleware.__class__.__name__, update_type))
result = None
else: else:
logger.error("Middleware: {} does not have post_process_{} method. Post process function was not executed.".format(middleware.__class__.__name__, update_type)) result = middleware.pre_process(message, data)
else: # We will break this loop if CancelUpdate is returned
middleware.post_process(message, data, handler_error) # Also, we will not run other middlewares
if isinstance(result, CancelUpdate):
return
elif isinstance(result, SkipHandler):
skip_handlers = True
if handlers and not(skip_handlers):
try:
for handler in handlers:
process_handler = self._test_message_handler(handler, message)
if not process_handler: continue
for i in inspect.signature(handler['function']).parameters:
params.append(i)
if len(params) == 1:
handler['function'](message)
elif "data" in params:
if len(params) == 2:
handler['function'](message, data)
elif len(params) == 3:
handler['function'](message, data=data, bot=self)
else:
logger.error("It is not allowed to pass data and values inside data to the handler. Check your handler: {}".format(handler['function']))
return
else:
data_copy = data.copy()
for key in list(data_copy):
# remove data from data_copy if handler does not accept it
if key not in params:
del data_copy[key]
if handler.get('pass_bot'):
data_copy["bot"] = self
if len(data_copy) > len(params) - 1: # remove the message parameter
logger.error("You are passing more parameters than the handler needs. Check your handler: {}".format(handler['function']))
return
handler["function"](message, **data_copy)
break
except Exception as e:
handler_error = e
if self.exception_handler:
self.exception_handler.handle(e)
else:
logging.error(str(e))
logger.debug("Exception traceback:\n%s", traceback.format_exc())
if middlewares:
for middleware in middlewares:
if middleware.update_sensitive:
if hasattr(middleware, f'post_process_{update_type}'):
getattr(middleware, f'post_process_{update_type}')(message, data, handler_error)
else:
logger.error("Middleware: {} does not have post_process_{} method. Post process function was not executed.".format(middleware.__class__.__name__, update_type))
else:
middleware.post_process(message, data, handler_error)
def _notify_command_handlers(self, handlers, new_messages, update_type): def _notify_command_handlers(self, handlers, new_messages, update_type):
""" """
Notifies command handlers. Notifies command handlers.
:param handlers: :param handlers: all created handlers
:param new_messages: :param new_messages: received messages to proceed
:param update_type: handler/update type (Update fields)
:return: :return:
""" """
if not(handlers) and not(self.use_class_middlewares): if not(handlers) and not(self.use_class_middlewares):
return return
if self.use_class_middlewares:
middlewares = self._check_middlewares(update_type)
else:
middlewares = None
for message in new_messages: for message in new_messages:
if not self.use_class_middlewares: self._exec_task(
for message_handler in handlers: self._run_middlewares_and_handler,
if self._test_message_handler(message_handler, message): message,
self._exec_task(message_handler['function'], message, pass_bot=message_handler['pass_bot'], task_type='handler') handlers=handlers,
break middlewares=middlewares,
else: update_type=update_type)
middleware = self._check_middleware(update_type)
self._exec_task(self._run_middlewares_and_handler, message, handlers=handlers, middlewares=middleware, update_type=update_type)
return

View File

@ -1,6 +1,5 @@
""" """
This file is used by TeleBot.run_webhooks() function. This file is used by TeleBot.run_webhooks() function.
Fastapi is required to run this script. Fastapi is required to run this script.
""" """
@ -15,15 +14,11 @@ try:
except ImportError: except ImportError:
fastapi_installed = False fastapi_installed = False
from telebot.types import Update from telebot.types import Update
from typing import Optional from typing import Optional
class SyncWebhookListener: class SyncWebhookListener:
def __init__(self, bot, def __init__(self, bot,
secret_token: str, host: Optional[str]="127.0.0.1", secret_token: str, host: Optional[str]="127.0.0.1",
@ -33,13 +28,13 @@ class SyncWebhookListener:
debug: Optional[bool]=False debug: Optional[bool]=False
) -> None: ) -> None:
""" """
Aynchronous implementation of webhook listener Synchronous implementation of webhook listener
for asynchronous version of telebot. for synchronous version of telebot.
Not supposed to be used manually by user. Not supposed to be used manually by user.
Use AsyncTeleBot.run_webhooks() instead. Use TeleBot.run_webhooks() instead.
:param bot: AsyncTeleBot instance. :param bot: TeleBot instance.
:type bot: telebot.async_telebot.AsyncTeleBot :type bot: telebot.TeleBot
:param secret_token: Telegram secret token :param secret_token: Telegram secret token
:type secret_token: str :type secret_token: str
@ -77,7 +72,8 @@ class SyncWebhookListener:
self._prepare_endpoint_urls() self._prepare_endpoint_urls()
def _check_dependencies(self): @staticmethod
def _check_dependencies():
if not fastapi_installed: if not fastapi_installed:
raise ImportError('Fastapi or uvicorn is not installed. Please install it via pip.') raise ImportError('Fastapi or uvicorn is not installed. Please install it via pip.')

View File

@ -41,7 +41,10 @@ class StateStorageBase:
def get_state(self, chat_id, user_id): def get_state(self, chat_id, user_id):
raise NotImplementedError raise NotImplementedError
def get_interactive_data(self, chat_id, user_id):
raise NotImplementedError
def save(self, chat_id, user_id, data): def save(self, chat_id, user_id, data):
raise NotImplementedError raise NotImplementedError

View File

@ -3,6 +3,7 @@ from telebot.storage.base_storage import StateStorageBase, StateContext
class StateMemoryStorage(StateStorageBase): class StateMemoryStorage(StateStorageBase):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__()
self.data = {} self.data = {}
# #
# {chat_id: {user_id: {'state': None, 'data': {}}, ...}, ...} # {chat_id: {user_id: {'state': None, 'data': {}}, ...}, ...}

View File

@ -5,8 +5,8 @@ import pickle
class StatePickleStorage(StateStorageBase): class StatePickleStorage(StateStorageBase):
# noinspection PyMissingConstructor
def __init__(self, file_path="./.state-save/states.pkl") -> None: def __init__(self, file_path="./.state-save/states.pkl") -> None:
super().__init__()
self.file_path = file_path self.file_path = file_path
self.create_dir() self.create_dir()
self.data = self.read() self.data = self.read()

View File

@ -16,6 +16,7 @@ class StateRedisStorage(StateStorageBase):
TeleBot(storage=StateRedisStorage()) TeleBot(storage=StateRedisStorage())
""" """
def __init__(self, host='localhost', port=6379, db=0, password=None, prefix='telebot_'): def __init__(self, host='localhost', port=6379, db=0, password=None, prefix='telebot_'):
super().__init__()
self.redis = ConnectionPool(host=host, port=port, db=db, password=password) self.redis = ConnectionPool(host=host, port=port, db=db, password=password)
#self.con = Redis(connection_pool=self.redis) -> use this when necessary #self.con = Redis(connection_pool=self.redis) -> use this when necessary
# #