remove staging to output copying logic from executor

Author: andrii-i

jupyter_scheduler/executors.py

@@ -14,7 +14,7 @@
 from jupyter_scheduler.models import DescribeJob, JobFeature, JobFile, Status
 from jupyter_scheduler.orm import Job, create_session
 from jupyter_scheduler.parameterize import add_parameters
-from jupyter_scheduler.utils import copy_directory, get_utc_timestamp
+from jupyter_scheduler.utils import get_utc_timestamp
 
 
 class ExecutionManager(ABC):
@@ -29,19 +29,11 @@ class ExecutionManager(ABC):
     _model = None
     _db_session = None
 
-    def __init__(
-        self,
-        job_id: str,
-        root_dir: str,
-        db_url: str,
-        staging_paths: Dict[str, str],
-        output_dir: str,
-    ):
+    def __init__(self, job_id: str, root_dir: str, db_url: str, staging_paths: Dict[str, str]):
         self.job_id = job_id
         self.staging_paths = staging_paths
         self.root_dir = root_dir
         self.db_url = db_url
-        self.output_dir = output_dir
 
     @property
     def model(self):
@@ -155,7 +147,6 @@ def execute(self):
                 output, _ = cls().from_notebook_node(nb)
                 with fsspec.open(self.staging_paths[output_format], "w", encoding="utf-8") as f:
                     f.write(output)
-            self.copy_staged_files_to_output()
 
     def add_side_effects_files(self, staging_dir):
         """Scan for side effect files potentially created after input file execution and update the job's packaged_files with these files"""
@@ -179,12 +170,6 @@ def add_side_effects_files(self, staging_dir):
                 )
                 session.commit()
 
-    # TODO: copy via downloader and remove this function or use this function and utilize return for putting side efects into packaged_files
-    def copy_staged_files_to_output(self):
-        """Copies snapshot of the original notebook and staged input files from the staging directory to the output directory and includes them into job_files."""
-        staging_dir = os.path.dirname(self.staging_paths["input"])
-        copy_directory(source_dir=staging_dir, destination_dir=self.output_dir)
-
     def supported_features(cls) -> Dict[JobFeature, bool]:
         return {
             JobFeature.job_name: True,

jupyter_scheduler/job_files_manager.py

@@ -21,9 +21,7 @@ async def copy_from_staging(self, job_id: str, redownload: Optional[bool] = Fals
         job = await ensure_async(self.scheduler.get_job(job_id, False))
         staging_paths = await ensure_async(self.scheduler.get_staging_paths(job))
         output_filenames = self.scheduler.get_job_filenames(job)
-        output_dir = self.scheduler.get_local_output_path(
-            input_filename=job.input_filename, job_id=job.job_id, root_dir_relative=True
-        )
+        output_dir = self.scheduler.get_local_output_path(model=job, root_dir_relative=True)
 
         p = Process(
             target=Downloader(

jupyter_scheduler/scheduler.py

@@ -319,9 +319,7 @@ def add_job_files(self, model: DescribeJob):
         mapping = self.environments_manager.output_formats_mapping()
         job_files = []
         output_filenames = self.get_job_filenames(model)
-        output_dir = self.get_local_output_path(
-            input_filename=model.input_filename, job_id=model.job_id, root_dir_relative=True
-        )
+        output_dir = self.get_local_output_path(model, root_dir_relative=True)
 
         for output_format in model.output_formats:
             filename = output_filenames[output_format]
@@ -369,13 +367,13 @@ def add_job_files(self, model: DescribeJob):
         )
 
     def get_local_output_path(
-        self, input_filename: str, job_id: str, root_dir_relative: Optional[bool] = False
+        self, model: DescribeJob, root_dir_relative: Optional[bool] = False
     ) -> str:
         """Returns the local output directory path
         where all the job files will be downloaded
         from the staging location.
         """
-        output_dir_name = create_output_directory(input_filename, job_id)
+        output_dir_name = create_output_directory(model.input_filename, model.job_id)
         if root_dir_relative:
             return os.path.relpath(
                 os.path.join(self.root_dir, self.output_directory, output_dir_name), self.root_dir
@@ -494,10 +492,6 @@ def create_job(self, model: CreateJob) -> str:
                     staging_paths=staging_paths,
                     root_dir=self.root_dir,
                     db_url=self.db_url,
-                    output_dir=self.get_local_output_path(
-                        input_filename=model.input_filename,
-                        job_id=job.job_id,
-                    ),
                 ).process
             )
             p.start()