def start_tasks()

in src/dispatch/cli.py [0:0]

• 27 lines of code
• 12 McCabe index (conditional complexity)

def start_tasks(tasks, exclude, eager):
    """Starts the scheduler."""
    import signal

    from dispatch.common.utils.cli import install_plugins
    from dispatch.scheduler import stop_scheduler

    install_plugins()

    if tasks:
        for task in scheduler.registered_tasks:
            if task["name"] not in tasks:
                scheduler.remove(task)

    if exclude:
        for task in scheduler.registered_tasks:
            if task["name"] in exclude:
                scheduler.remove(task)

    if eager:
        for task in tasks:
            for r_task in scheduler.registered_tasks:
                if task == r_task["name"]:
                    click.secho(f"Eagerly running: {task}", fg="blue")
                    r_task["func"]()
                    break
            else:
                click.secho(f"A scheduled task/job named {task} does not exist", fg="red")

    # registers a handler to stop future scheduling when encountering sigterm
    signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT)
    for s in signals:
        signal.signal(s, stop_scheduler)

    click.secho("Starting scheduler...", fg="blue")
    scheduler.start()