""" This file contains the base MEOW handler defintion. This should be inherited from for all handler instances. Author(s): David Marchant """ from typing import Any, Tuple, Dict from meow_base.core.vars import VALID_CHANNELS, \ VALID_HANDLER_NAME_CHARS, get_drt_imp_msg from meow_base.functionality.validation import check_implementation, \ valid_string from meow_base.functionality.naming import generate_handler_id class BaseHandler: # An identifier for a handler within the runner. Can be manually set in # the constructor, or autogenerated if no name provided. name:str # A channel for sending messages to the runner. Note that this will be # overridden by a MeowRunner, if a handler instance is passed to it, and so # does not need to be initialised within the handler itself. to_runner: VALID_CHANNELS # Directory where queued jobs are initially written to. Note that this # will be overridden by a MeowRunner, if a handler instance is passed to # it, and so does not need to be initialised within the handler itself. job_queue_dir:str def __init__(self, name:str='')->None: """BaseHandler Constructor. This will check that any class inheriting from it implements its validation functions.""" check_implementation(type(self).handle, BaseHandler) check_implementation(type(self).valid_handle_criteria, BaseHandler) if not name: name = generate_handler_id() self._is_valid_name(name) self.name = name def __new__(cls, *args, **kwargs): """A check that this base class is not instantiated itself, only inherited from""" if cls is BaseHandler: msg = get_drt_imp_msg(BaseHandler) raise TypeError(msg) return object.__new__(cls) def _is_valid_name(self, name:str)->None: """Validation check for 'name' variable from main constructor. Is automatically called during initialisation. This does not need to be overridden by child classes.""" valid_string(name, VALID_HANDLER_NAME_CHARS) def valid_handle_criteria(self, event:Dict[str,Any])->Tuple[bool,str]: """Function to determine given an event defintion, if this handler can process it or not. Must be implemented by any child process.""" pass def handle(self, event:Dict[str,Any])->None: """Function to handle a given event. Must be implemented by any child process.""" pass