Method process_new_event

eventhandlers/SchedulerConsumer.py:73–111

Receives a trigger event a fi queueclient = QueueClient()nds any actions matching the event trigger for action execution. :return:

(channel, method, properties, msg)

    @staticmethod
    def process_new_event(channel, method, properties, msg):
        """
            Receives a trigger event a fi
            queueclient = QueueClient()nds any actions
            matching the event trigger for action
            execution.
        :return:
        """
        logger.debug(f'New Scheduler Message:  {msg}')
        msg_data = json.loads(msg)
        log = regular_log.default()
        if msg_data.get('workflow_id') is None:
            log['error']['workflow_id'] = f'Message must contain workflow_id. Message is: {msg_data}'
        if msg_data.get('project_id') is None:
            log['error']['project_id'] = f'Message must contain project_id. Message is: {msg_data}'
        if msg_data.get('action') is None:
            log['error']['action'] = f'Message most contain an action. Message is: {msg_data}'
        if msg_data.get('cron_expression') is None:
            log['error']['cron_expression'] = f'Message must contain a cron_expression. Message is: {msg_data}'


        if regular_log.log_has_error(log):
            logger.error(f'Error processing jobs message: {log}')
            return

        workflow_id = msg_data.get('workflow_id')
        project_id = msg_data.get('project_id')
        action = msg_data.get('action')
        cron_expression = msg_data.get('cron_expression')
        logger.debug(f'Processing Scheduler event: {msg}')

        with session_scope_threaded() as session:
            if action == 'add':
                diffgram_scheduler.add_job(job_id = workflow_id, cron_expr = cron_expression, func = trigger_workflow, args = [workflow_id, project_id])
            elif action == 'remove':
                diffgram_scheduler.remove_job(job_id = workflow_id)
            else:
                logger.warning(f'Scheduler Consumer: Unknown action type "{action}"')
            logger.debug(f'Scheduler event processed successfully. {msg}')