Abort build and print reports on KeyboardInterrupt. (#25)

Author: tobiasraabe

.gitattributes

@@ -18,7 +18,7 @@ jobs:
       fail-fast: false
       matrix:
         os: ['ubuntu-latest', 'macos-latest', 'windows-latest']
-        python-version: ['3.6', '3.7', '3.8']
+        python-version: ['3.7', '3.8', '3.9']
 
     steps:
       - uses: actions/checkout@v2

.github/workflows/continuous-integration-workflow.yml

@@ -13,6 +13,7 @@ all releases are available on `PyPI <https://pypi.org/project/pytask-parallel>`_
 - :gh:`19` adds ``conda-forge`` to the ``README.rst``.
 - :gh:`22` add note that the debugger cannot be used together with pytask-parallel.
 - :gh:`24` replaces versioneer with setuptools-scm.
+- :gh:`25` aborts build and prints reports on ``KeyboardInterrupt``.
 
 
 0.0.8 - 2021-03-05

CHANGES.rst

@@ -5,7 +5,7 @@ channels:
   - nodefaults
 
 dependencies:
-  - python
+  - python=3.6
   - pip
   - setuptools_scm
   - toml
@@ -37,7 +37,6 @@ dependencies:
   - pdbpp
   - pre-commit
   - pytest-cov
-  - versioneer
 
   # Documentation
   - numpydoc

environment.yml

@@ -41,78 +41,81 @@ def pytask_execute_build(session):
 
             while session.scheduler.is_active():
 
-                newly_collected_reports = []
-                n_new_tasks = session.config["n_workers"] - len(running_tasks)
-
-                if n_new_tasks >= 1:
-                    ready_tasks = list(session.scheduler.get_ready(n_new_tasks))
-                else:
-                    ready_tasks = []
-
-                for task_name in ready_tasks:
-                    task = session.dag.nodes[task_name]["task"]
-                    session.hook.pytask_execute_task_log_start(
-                        session=session, task=task
-                    )
-                    try:
-                        session.hook.pytask_execute_task_setup(
-                            session=session, task=task
-                        )
-                    except Exception:
-                        report = ExecutionReport.from_task_and_exception(
-                            task, sys.exc_info()
-                        )
-                        newly_collected_reports.append(report)
-                        session.scheduler.done(task_name)
+                try:
+                    newly_collected_reports = []
+                    n_new_tasks = session.config["n_workers"] - len(running_tasks)
+
+                    if n_new_tasks >= 1:
+                        ready_tasks = list(session.scheduler.get_ready(n_new_tasks))
                     else:
-                        running_tasks[task_name] = session.hook.pytask_execute_task(
-                            session=session, task=task
-                        )
+                        ready_tasks = []
 
-                for task_name in list(running_tasks):
-                    future = running_tasks[task_name]
-                    if future.done() and future.exception() is not None:
+                    for task_name in ready_tasks:
                         task = session.dag.nodes[task_name]["task"]
-                        exception = future.exception()
-                        newly_collected_reports.append(
-                            ExecutionReport.from_task_and_exception(
-                                task, (type(exception), exception, None)
-                            )
+                        session.hook.pytask_execute_task_log_start(
+                            session=session, task=task
                         )
-                        running_tasks.pop(task_name)
-                        session.scheduler.done(task_name)
-                    elif future.done() and future.exception() is None:
-                        task = session.dag.nodes[task_name]["task"]
                         try:
-                            session.hook.pytask_execute_task_teardown(
+                            session.hook.pytask_execute_task_setup(
                                 session=session, task=task
                             )
                         except Exception:
                             report = ExecutionReport.from_task_and_exception(
                                 task, sys.exc_info()
                             )
+                            newly_collected_reports.append(report)
+                            session.scheduler.done(task_name)
                         else:
-                            report = ExecutionReport.from_task(task)
+                            running_tasks[task_name] = session.hook.pytask_execute_task(
+                                session=session, task=task
+                            )
+
+                    for task_name in list(running_tasks):
+                        future = running_tasks[task_name]
+                        if future.done() and future.exception() is not None:
+                            task = session.dag.nodes[task_name]["task"]
+                            exception = future.exception()
+                            newly_collected_reports.append(
+                                ExecutionReport.from_task_and_exception(
+                                    task, (type(exception), exception, None)
+                                )
+                            )
+                            running_tasks.pop(task_name)
+                            session.scheduler.done(task_name)
+                        elif future.done() and future.exception() is None:
+                            task = session.dag.nodes[task_name]["task"]
+                            try:
+                                session.hook.pytask_execute_task_teardown(
+                                    session=session, task=task
+                                )
+                            except Exception:
+                                report = ExecutionReport.from_task_and_exception(
+                                    task, sys.exc_info()
+                                )
+                            else:
+                                report = ExecutionReport.from_task(task)
+
+                            running_tasks.pop(task_name)
+                            newly_collected_reports.append(report)
+                            session.scheduler.done(task_name)
+                        else:
+                            pass
+
+                    for report in newly_collected_reports:
+                        session.hook.pytask_execute_task_process_report(
+                            session=session, report=report
+                        )
+                        session.hook.pytask_execute_task_log_end(
+                            session=session, task=task, report=report
+                        )
+                        reports.append(report)
 
-                        running_tasks.pop(task_name)
-                        newly_collected_reports.append(report)
-                        session.scheduler.done(task_name)
+                    if session.should_stop:
+                        break
                     else:
-                        pass
-
-                for report in newly_collected_reports:
-                    session.hook.pytask_execute_task_process_report(
-                        session=session, report=report
-                    )
-                    session.hook.pytask_execute_task_log_end(
-                        session=session, task=task, report=report
-                    )
-                    reports.append(report)
-
-                if session.should_stop:
+                        time.sleep(session.config["delay"])
+                except KeyboardInterrupt:
                     break
-                else:
-                    time.sleep(session.config["delay"])
 
         return True
 

src/pytask_parallel/execute.py

@@ -1,11 +1,11 @@
 """Contains code relevant to logging."""
-import click
 from _pytask.config import hookimpl
+from _pytask.console import console
 
 
 @hookimpl(trylast=True)
 def pytask_log_session_header(session):
     """Add a note for how many workers are spawned."""
     n_workers = session.config["n_workers"]
     if n_workers > 1:
-        click.echo(f"Started {n_workers} workers.")
+        console.print(f"Started {n_workers} workers.")