Fixing upload and prerparations for it

preprocessor/preprocessor/local.py

@@ -27,6 +27,7 @@ class Uploader(Base):
         paths = local_path if isinstance(local_path, List) else [local_path]
         remote_paths = [
             os.path.join(
+                self.storage_path,
                 remote_dir,
                 os.path.basename(path)
             )
@@ -34,6 +35,7 @@ class Uploader(Base):
         ]
 
         for local_path, remote_path in zip(paths, remote_paths):
+            os.makedirs(os.path.dirname(remote_path), exist_ok=True)
             shutil.copy2(local_path, remote_path)
 
         return remote_paths

preprocessor/preprocessor/preprocess.py

@@ -3,6 +3,7 @@ from tempfile import TemporaryDirectory
 import itertools
 import importlib
 import logging
+import shutil
 
 from .transfer import get_downloader, get_uploader
 from .archive import unpack_files
@@ -13,6 +14,21 @@ logger = logging.getLogger(__name__)
 
 # -----------------------------------------------------------------------------
 
+
+def copy_files(source, target):
+    for item in os.listdir(source):
+        print(item)
+        src_path = os.path.join(source, item)
+        dst_path = os.path.join(target, item)
+        if os.path.isdir(src_path):
+            shutil.copytree(
+                src_path,
+                dst_path
+            )
+        else:
+            shutil.copy(src_path, dst_path)
+
+
 def custom_preprocessor(source_dir, target_dir, path, args=None, kwargs=None):
     """ Preprocessing step for a custom preprocessing.
     """
@@ -70,6 +86,7 @@ def preprocess_file(config: dict, file_path: os.PathLike):
         os.chdir(tempdir)
         os.mkdir('download')
         os.mkdir('unpack')
+        os.mkdir('upload')
 
         # get the Downloader for the configured source archive to download the given source file
         source_config = config['source']
@@ -124,6 +141,8 @@ def preprocess_file(config: dict, file_path: os.PathLike):
 
             previous_step = step
 
+        # copy files from previous step directory to upload directory
+        copy_files(previous_step, 'upload')
 
         # get an uploader for the finalized images
         target_config = config['target']
@@ -131,5 +150,11 @@ def preprocess_file(config: dict, file_path: os.PathLike):
             target_config['type'], target_config.get('args'), target_config.get('kwargs')
         )
 
-        # TODO send all files in the output directory to the target storage
-        uploader.upload()
\ No newline at end of file
+        upload_filenames = [
+            os.path.join(dirpath, filename)
+            for dirpath, _, filenames in os.walk('upload')
+            for filename in filenames
+        ]
+
+        # send all files in the upload directory to the target storage
+        uploader.upload(upload_filenames, file_path)

preprocessor/preprocessor/transfer.py

 from . import swift
 from . import local
 
+from .abc import Downloader, Uploader
 
-def get_downloader(type_name, args, kwargs):
+
+def get_downloader(type_name, args, kwargs) -> Downloader:
     if type_name == 'swift':
         return swift.Downloader(*args or [], **kwargs or {})
     elif type_name == 'local':
@@ -11,10 +13,10 @@ def get_downloader(type_name, args, kwargs):
     raise Exception('Downloader type %s is not supported' % type_name)
 
 
-def get_uploader(uploader_config):
-    if uploader_config['type'] == 'swift':
-        return swift.Uploader(**uploader_config['kwargs'])
-    elif uploader_config['type'] == 'local':
-        return local.Uploader(**uploader_config['kwargs'])
+def get_uploader(type_name, args, kwargs) -> Uploader:
+    if type_name == 'swift':
+        return swift.Uploader(*args or [], **kwargs or {})
+    elif type_name == 'local':
+        return local.Uploader(*args or [], **kwargs or {})
 
-    raise Exception('Uploader type %s is not supported' % uploader_config['type'])
+    raise Exception('Uploader type %s is not supported' % type_name)