def _stop_job()

in dyno/app/api/control.py [0:0]

• 10 lines of code
• 3 McCabe index (conditional complexity)

def _stop_job(job: str) -> None:
    """
    Helper function for stopping a job

    Note
    ----
    This function can only be called directly and is
    not accessible via HTTP.

    Parameters
    ----------
    job : str
        The job to stop

    Returns
    -------
    None

    Examples
    --------
    >>> _stop_job('opbeans-python')


    """
    s = socketio.Client()
    s.emit('service_state', {'data': {job: 'stop'}})
    if job in JOB_MANAGER:
        p = JOB_MANAGER[job]
        os.killpg(os.getpgid(p.pid), signal.SIGTERM)
        JOB_MANAGER[job] = None
    if job in JOB_STATUS:
        j = JOB_STATUS[job]
        j['running'] = False