Repository: spotify/luigi
File: luigi/worker.py (Python)

Code before:
    def _get_work_task_id(self, get_work_response):
        if get_work_response['task_id'] is not None:
            return get_work_response['task_id']
        elif 'batch_id' in get_work_response:
            try:
                task = load_task(
                    module=get_work_response.get('task_module'),
                    task_name=get_work_response['task_family'],
                    params_str=get_work_response['task_params'],
                )
            except Exception as ex:
                batch_task_ids = get_work_response['batch_task_ids']
                msg = 'Cannot load batch task for %s' % ', '.join(batch_task_ids)
                logger.exception(msg)
                subject = 'Luigi: %s' % msg
                error_message = notifications.wrap_traceback(ex)
                notifications.send_error_email(subject, error_message)
                for batch_task_id in batch_task_ids:
                    self._scheduler.add_task(
                        worker=self._id, status=FAILED, task_id=batch_task_id, expl=error_message)
                self.run_succeeded = False
                return None

            self._scheduler.add_task(
                worker=self._id,
                task_id=task.task_id,
                module=get_work_response.get('task_module'),
                family=get_work_response['task_family'],
                params=task.to_str_params(),
                status=RUNNING,
                batch_id=get_work_response['batch_id'],
            )
            return task.task_id

    def _get_work(self):
        if self._stop_requesting_work:
            return None, 0, 0, 0, WORKER_STATE_DISABLED
        logger.debug("Asking scheduler for work...")
        r = self._scheduler.get_work(
            worker=self._id,
            host=self.host,
            assistant=self._assistant,
            current_tasks=list(self._running_tasks.keys()),
        )
        n_pending_tasks = r['n_pending_tasks']
        running_tasks = r['running_tasks']
        n_unique_pending = r['n_unique_pending']
        # TODO: For a tiny amount of time (a month?) we'll keep forwards compatibility
        # That is you can user a newer client than server (Sep 2016)
        worker_state = r.get('worker_state', WORKER_STATE_ACTIVE)  # state according to server!
        task_id = self._get_work_task_id(r)

Review comment (bug):
Since you catch such a broad exception, the code isn't self-documenting. I'm sure your use-case is fine, but can you add a comment within the `except` block of what typically goes wrong and makes us end up here?


Code after:
    def _add_worker(self):
        self._worker_info.append(('first_task', self._first_task))
        self._scheduler.add_worker(self._id, self._worker_info)

    def _log_remote_tasks(self, running_tasks, n_pending_tasks, n_unique_pending):
        logger.debug("Done")
        logger.debug("There are no more tasks to run at this time")
        if running_tasks:
            for r in running_tasks:
                logger.debug('%s is currently run by worker %s', r['task_id'], r['worker'])
        elif n_pending_tasks:
            logger.debug("There are %s pending tasks possibly being run by other workers", n_pending_tasks)
            if n_unique_pending:
                logger.debug("There are %i pending tasks unique to this worker", n_unique_pending)

    def _get_work_task_id(self, get_work_response):
        if get_work_response['task_id'] is not None:
            return get_work_response['task_id']
        elif 'batch_id' in get_work_response:
            try:
                task = load_task(
                    module=get_work_response.get('task_module'),
                    task_name=get_work_response['task_family'],
                    params_str=get_work_response['task_params'],
                )
            except Exception as ex:
                self._handle_task_load_error(ex, get_work_response['batch_task_ids'])
                self.run_succeeded = False
                return None

            self._scheduler.add_task(
                worker=self._id,
                task_id=task.task_id,
                module=get_work_response.get('task_module'),
                family=get_work_response['task_family'],
                params=task.to_str_params(),
                status=RUNNING,
                batch_id=get_work_response['batch_id'],
            )
            return task.task_id

    def _get_work(self):
        if self._stop_requesting_work:
            return None, 0, 0, 0, WORKER_STATE_DISABLED
        logger.debug("Asking scheduler for work...")
        r = self._scheduler.get_work(
            worker=self._id,
            host=self.host,
            assistant=self._assistant,
            current_tasks=list(self._running_tasks.keys()),
        )