def execute()

in pygenie/jobs/core.py [0:0]

• 34 lines of code
• 13 McCabe index (conditional complexity)

    def execute(self, retry=False, force=False, override_existing=False, catch_signal=False, **kwargs):
        """
        Send the job to Genie and execute.

        Example:
            >>> running_job = GenieJob().execute()

        Args:
            retry (bool, optional): If True, will check to see if a previous job
                with the same job id has previously executed and generate a new
                job id until finds a job id to a running job, a successful job, or
                a brand new job (Default: False).
            force (bool, optional): If True, will do the same thing as
                retry=True, but will generate a new job id even if there is
                a successful execution (Default: False).
            override_existing (bool, optional): If True, will do the same thing as
                force=True, but will generate a new job id and kill the previous
                running executions (Default: False).
            catch_signal (bool, optional): If True, will add signal handlers to
                kill the running job for SIGINT, SIGTERM, and SIGABRT
                (Default: False).

        Returns:
            :py:class:`RunningJob`: A running job object.

        Raises:
            ValueError: If `override_existing` is True without `force` being True.
        """

        if not force and override_existing:
            raise ValueError("override_existing cannot be True without force=True")

        if catch_signal:
            import signal
            import sys
            def sig_handler(signum, frame):
                logger.warning("caught signal %s", signum)
                try:
                    if running_job.job_id:
                        logger.warning("killing job id %s", running_job.job_id)
                        response = running_job.kill()
                        response.raise_for_status()
                except Exception:
                    pass
                finally:
                    sys.exit(1)
            signal.signal(signal.SIGINT, sig_handler)
            signal.signal(signal.SIGTERM, sig_handler)
            signal.signal(signal.SIGABRT, sig_handler)

        if retry or force:
            uid = self._job_id
            try:
                # below uid should be uid of job with one of following status:
                #     - new
                #     - (if override_existing=False) running
                #     - (if force=False) successful
                uid = generate_job_id(uid,
                                      return_success=not force,
                                      override_existing=override_existing,
                                      conf=self._conf)
                # new uid will raise and be handled in the except block
                # assigning to running_job variable for killing on signal
                running_job = reattach_job(uid, conf=self._conf)
                return running_job
            except GenieJobNotFoundError:
                self.job_id(uid)

        global execute_job  # set in main __init__.py to avoid circular imports
        # execute_job imports jobs, jobs need to import execute_job
        # assigning to running_job variable for killing on signal
        running_job = execute_job(self, **kwargs)
        return running_job