refactor output files creation logic into a separate function for clarity

Author: andrii-i

jupyter_scheduler/executors.py

@@ -11,13 +11,9 @@
 import nbformat
 from nbconvert.preprocessors import CellExecutionError, ExecutePreprocessor
 
-from jupyter_scheduler.download_manager import (
-    DescribeDownload,
-    Download,
-    initiate_download_standalone,
-)
+from jupyter_scheduler.download_manager import initiate_download_standalone
 from jupyter_scheduler.models import DescribeJob, JobFeature, Status
-from jupyter_scheduler.orm import Job, create_session, generate_uuid
+from jupyter_scheduler.orm import Job, create_session
 from jupyter_scheduler.parameterize import add_parameters
 from jupyter_scheduler.utils import get_utc_timestamp
 
@@ -155,11 +151,7 @@ def execute(self):
             raise e
         finally:
             self.add_side_effects_files(staging_dir)
-            for output_format in job.output_formats:
-                cls = nbconvert.get_exporter(output_format)
-                output, _ = cls().from_notebook_node(nb)
-                with fsspec.open(self.staging_paths[output_format], "w", encoding="utf-8") as f:
-                    f.write(output)
+            self.create_output_files(job, nb)
             with self.db_session() as session:
                 initiate_download_standalone(
                     job_id=job.job_id,
@@ -168,7 +160,7 @@ def execute(self):
                     redownload=True,
                 )
 
-    def add_side_effects_files(self, staging_dir):
+    def add_side_effects_files(self, staging_dir: str):
         """Scan for side effect files potentially created after input file execution and update the job's packaged_files with these files"""
         input_notebook = os.path.relpath(self.staging_paths["input"])
         new_files_set = set()
@@ -190,6 +182,13 @@ def add_side_effects_files(self, staging_dir):
                 )
                 session.commit()
 
+    def create_output_files(self, job: DescribeJob, notebook_node):
+        for output_format in job.output_formats:
+            cls = nbconvert.get_exporter(output_format)
+            output, _ = cls().from_notebook_node(notebook_node)
+            with fsspec.open(self.staging_paths[output_format], "w", encoding="utf-8") as f:
+                f.write(output)
+
     def supported_features(cls) -> Dict[JobFeature, bool]:
         return {
             JobFeature.job_name: True,