diff --git a/aiida/cmdline/commands/cmd_export.py b/aiida/cmdline/commands/cmd_export.py
index b282d8a78a..add1f9641e 100644
--- a/aiida/cmdline/commands/cmd_export.py
+++ b/aiida/cmdline/commands/cmd_export.py
@@ -79,7 +79,9 @@ def inspect(archive, version, data, meta_data):
 @options.COMPUTERS()
 @options.GROUPS()
 @options.NODES()
-@options.ARCHIVE_FORMAT()
+@options.ARCHIVE_FORMAT(
+    type=click.Choice(['zip', 'zip-uncompressed', 'zip-lowmemory', 'tar.gz', 'null']),
+)
 @options.FORCE(help='overwrite output file if it already exists')
 @click.option(
     '-v',
@@ -101,6 +103,14 @@ def inspect(archive, version, data, meta_data):
     show_default=True,
     help='Include or exclude comments for node(s) in export. (Will also export extra users who commented).'
 )
+# will only be useful when moving to a new archive format, that does not store all data in memory
+# @click.option(
+#     '-b',
+#     '--batch-size',
+#     default=1000,
+#     type=int,
+#     help='Batch database query results in sub-collections to reduce memory usage.'
+# )
 @decorators.with_dbenv()
 def create(
     output_file, codes, computers, groups, nodes, archive_format, force, input_calc_forward, input_work_forward,
@@ -115,6 +125,7 @@ def create(
     their provenance, according to the rules outlined in the documentation.
     You can modify some of those rules using options of this command.
     """
+    # pylint: disable=too-many-branches
     from aiida.common.log import override_log_formatter_context
     from aiida.common.progress_reporter import set_progress_bar_tqdm, set_progress_reporter
     from aiida.tools.importexport import export, ExportFileFormat, EXPORT_LOGGER
@@ -143,17 +154,22 @@ def create(
         'call_work_backward': call_work_backward,
         'include_comments': include_comments,
         'include_logs': include_logs,
-        'overwrite': force
+        'overwrite': force,
     }
 
     if archive_format == 'zip':
         export_format = ExportFileFormat.ZIP
-        kwargs.update({'use_compression': True})
+        kwargs.update({'writer_init': {'use_compression': True}})
     elif archive_format == 'zip-uncompressed':
         export_format = ExportFileFormat.ZIP
-        kwargs.update({'use_compression': False})
+        kwargs.update({'writer_init': {'use_compression': False}})
+    elif archive_format == 'zip-lowmemory':
+        export_format = ExportFileFormat.ZIP
+        kwargs.update({'writer_init': {'cache_zipinfo': True}})
     elif archive_format == 'tar.gz':
         export_format = ExportFileFormat.TAR_GZIPPED
+    elif archive_format == 'null':
+        export_format = 'null'
 
     if verbosity in ['DEBUG', 'INFO']:
         set_progress_bar_tqdm(leave=(verbosity == 'DEBUG'))
@@ -237,7 +253,10 @@ def migrate(input_file, output_file, force, silent, in_place, archive_format, ve
     except Exception as error:  # pylint: disable=broad-except
         if verbosity == 'DEBUG':
             raise
-        echo.echo_critical(f'failed to migrate the archive file (use `--verbosity DEBUG` to see traceback): {error}')
+        echo.echo_critical(
+            'failed to migrate the archive file (use `--verbosity DEBUG` to see traceback): '
+            f'{error.__class__.__name__}:{error}'
+        )
 
     if verbosity in ['DEBUG', 'INFO']:
         echo.echo_success(f'migrated the archive to version {version}')
diff --git a/aiida/cmdline/commands/cmd_import.py b/aiida/cmdline/commands/cmd_import.py
index 70a88ba26c..251006375e 100644
--- a/aiida/cmdline/commands/cmd_import.py
+++ b/aiida/cmdline/commands/cmd_import.py
@@ -207,7 +207,7 @@ def _import_archive(archive: str, web_based: bool, import_kwargs: dict, try_migr
         archive_path = archive
 
         if web_based:
-            echo.echo_info(f'downloading archive {archive}')
+            echo.echo_info(f'downloading archive: {archive}')
             try:
                 response = urllib.request.urlopen(archive)
             except Exception as exception:
@@ -216,6 +216,7 @@ def _import_archive(archive: str, web_based: bool, import_kwargs: dict, try_migr
             archive_path = temp_folder.get_abs_path('downloaded_archive.zip')
             echo.echo_success('archive downloaded, proceeding with import')
 
+        echo.echo_info(f'starting import: {archive}')
         try:
             import_data(archive_path, **import_kwargs)
         except IncompatibleArchiveVersionError as exception:
diff --git a/aiida/tools/importexport/archive/common.py b/aiida/tools/importexport/archive/common.py
index 83e32c728b..583115fbf5 100644
--- a/aiida/tools/importexport/archive/common.py
+++ b/aiida/tools/importexport/archive/common.py
@@ -15,17 +15,13 @@
 from pathlib import Path
 import tarfile
 from types import TracebackType
-from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Type, Union
+from typing import Any, Dict, List, Optional, Tuple, Type, Union
 import zipfile
 
 from aiida.common import json  # handles byte dumps
 from aiida.common.log import AIIDA_LOGGER
-from aiida.tools.importexport.common.exceptions import CorruptArchive
 
-__all__ = (
-    'ArchiveMetadata', 'detect_archive_type', 'null_callback', 'read_file_in_zip', 'read_file_in_tar',
-    'safe_extract_zip', 'safe_extract_tar', 'CacheFolder'
-)
+__all__ = ('ArchiveMetadata', 'detect_archive_type', 'null_callback', 'CacheFolder')
 
 ARCHIVE_LOGGER = AIIDA_LOGGER.getChild('archive')
 
@@ -47,7 +43,7 @@ class ArchiveMetadata:
     # optional data
     graph_traversal_rules: Optional[Dict[str, bool]] = dataclasses.field(default=None)
     # Entity type -> UUID list
-    entities_starting_set: Optional[Dict[str, Set[str]]] = dataclasses.field(default=None)
+    entities_starting_set: Optional[Dict[str, List[str]]] = dataclasses.field(default=None)
     include_comments: Optional[bool] = dataclasses.field(default=None)
     include_logs: Optional[bool] = dataclasses.field(default=None)
     # list of migration event notifications
@@ -80,229 +76,6 @@ def detect_archive_type(in_path: str) -> str:
     )
 
 
-def read_file_in_zip(filepath: str, path: str) -> str:
-    """Read a text based file from inside a zip file and return its content.
-
-    :param filepath: the path to the zip file
-    :param path: the relative path within the zip file
-
-    """
-    try:
-        return zipfile.ZipFile(filepath, 'r', allowZip64=True).read(path).decode('utf8')
-    except zipfile.BadZipfile as error:
-        raise CorruptArchive(f'The input file cannot be read: {error}')
-    except KeyError:
-        raise CorruptArchive(f'required file {path} is not included')
-
-
-def read_file_in_tar(filepath: str, path: str) -> str:
-    """Read a text based file from inside a tar file and return its content.
-
-    :param filepath: the path to the tar file
-    :param path: the relative path within the tar file
-
-    """
-    try:
-        with tarfile.open(filepath, 'r:*', format=tarfile.PAX_FORMAT) as handle:
-            result = handle.extractfile(path)
-            if result is None:
-                raise CorruptArchive(f'required file `{path}` is not included')
-            output = result.read()
-            if isinstance(output, bytes):
-                return output.decode('utf8')
-    except tarfile.ReadError:
-        raise ValueError('The input file format is not valid (not a tar file)')
-    except (KeyError, AttributeError):
-        raise CorruptArchive(f'required file `{path}` is not included')
-
-
-def _get_filter(only_prefix: Iterable[str], ignore_prefix: Iterable[str]) -> Callable[[str], bool]:
-    """Create filter for members to extract.
-
-    :param only_prefix: Extract only internal paths starting with these prefixes
-    :param ignore_prefix: Ignore internal paths starting with these prefixes
-
-    """
-    if only_prefix:
-
-        def _filter(name):
-            return any(name.startswith(prefix) for prefix in only_prefix
-                       ) and all(not name.startswith(prefix) for prefix in ignore_prefix)
-    else:
-
-        def _filter(name):
-            return all(not name.startswith(prefix) for prefix in ignore_prefix)
-
-    return _filter
-
-
-def safe_extract_zip(
-    in_path: Union[str, Path],
-    out_path: Union[str, Path],
-    *,
-    only_prefix: Iterable[str] = (),
-    ignore_prefix: Iterable[str] = ('..', '/'),
-    callback: Callable[[str, Any], None] = null_callback,
-    callback_description: str = 'Extracting zip files'
-):
-    """Safely extract a zip file
-
-    :param in_path: Path to extract from
-    :param out_path: Path to extract to
-    :param only_prefix: Extract only internal paths starting with these prefixes
-    :param ignore_prefix: Ignore internal paths starting with these prefixes
-    :param callback: a callback to report on the process, ``callback(action, value)``,
-        with the following callback signatures:
-
-        - ``callback('init', {'total': <int>, 'description': <str>})``,
-            to signal the start of a process, its total iterations and description
-        - ``callback('update', <int>)``,
-            to signal an update to the process and the number of iterations to progress
-
-    :param callback_description: the description to return in the callback
-
-    :raises `~aiida.tools.importexport.common.exceptions.CorruptArchive`: if the file cannot be read
-
-    """
-    _filter = _get_filter(only_prefix, ignore_prefix)
-    try:
-        with zipfile.ZipFile(in_path, 'r', allowZip64=True) as handle:
-            callback('init', {'total': 1, 'description': 'Gathering list of files to extract from zip'})
-            members = [name for name in handle.namelist() if _filter(name)]
-            callback('init', {'total': len(members), 'description': callback_description})
-            for membername in members:
-                callback('update', 1)
-                handle.extract(path=os.path.abspath(out_path), member=membername)
-    except zipfile.BadZipfile as error:
-        raise CorruptArchive(f'The input file cannot be read: {error}')
-
-
-def safe_extract_tar(
-    in_path: Union[str, Path],
-    out_path: Union[str, Path],
-    *,
-    only_prefix: Iterable[str] = (),
-    ignore_prefix: Iterable[str] = ('..', '/'),
-    callback: Callable[[str, Any], None] = null_callback,
-    callback_description: str = 'Extracting tar files'
-):
-    """Safely extract a tar file
-
-    :param in_path: Path to extract from
-    :param out_path: Path to extract to
-    :param only_prefix: Extract only internal paths starting with these prefixes
-    :param ignore_prefix: Ignore internal paths starting with these prefixes
-    :param callback: a callback to report on the process, ``callback(action, value)``,
-        with the following callback signatures:
-
-        - ``callback('init', {'total': <int>, 'description': <str>})``,
-            to signal the start of a process, its total iterations and description
-        - ``callback('update', <int>)``,
-            to signal an update to the process and the number of iterations to progress
-
-    :param callback_description: the description to return in the callback
-
-    :raises `~aiida.tools.importexport.common.exceptions.CorruptArchive`: if the file cannot be read
-
-    """
-    _filter = _get_filter(only_prefix, ignore_prefix)
-    try:
-        with tarfile.open(in_path, 'r:*', format=tarfile.PAX_FORMAT) as handle:
-            callback('init', {'total': 1, 'description': 'Computing tar objects to extract'})
-            members = [m for m in handle.getmembers() if _filter(m.name)]
-            callback('init', {'total': len(members), 'description': callback_description})
-            for member in members:
-                callback('update', 1)
-                if member.isdev():
-                    # safety: skip if character device, block device or FIFO
-                    msg = f'WARNING, device found inside the tar file: {member.name}'
-                    ARCHIVE_LOGGER.warning(msg)
-                    continue
-                if member.issym() or member.islnk():
-                    # safety: skip symlinks
-                    msg = f'WARNING, symlink found inside the tar file: {member.name}'
-                    ARCHIVE_LOGGER.warning(msg)
-                    continue
-                handle.extract(path=os.path.abspath(out_path), member=member)
-    except tarfile.ReadError as error:
-        raise CorruptArchive(f'The input file cannot be read: {error}')
-
-
-def compress_folder_zip(
-    in_path: Union[str, Path],
-    out_path: Union[str, Path],
-    *,
-    compression: int = zipfile.ZIP_DEFLATED,
-    callback: Callable[[str, Any], None] = null_callback,
-    callback_description: str = 'Compressing objects as zip'
-):
-    """Compress a folder as a zip file
-
-    :param in_path: Path to compress
-    :param out_path: Path to compress to
-    :param compression: the compression type (see zipfile module)
-    :param callback: a callback to report on the process, ``callback(action, value)``,
-        with the following callback signatures:
-
-        - ``callback('init', {'total': <int>, 'description': <str>})``,
-            to signal the start of a process, its total iterations and description
-        - ``callback('update', <int>)``,
-            to signal an update to the process and the number of iterations to progress
-
-    :param callback_description: the description to return in the callback
-
-    """
-    callback('init', {'total': 1, 'description': 'Computing objects to compress'})
-    count = 0
-    for _, dirnames, filenames in os.walk(in_path):
-        count += len(dirnames) + len(filenames)
-    callback('init', {'total': count, 'description': callback_description})
-    with zipfile.ZipFile(out_path, mode='w', compression=compression, allowZip64=True) as archive:
-        for dirpath, dirnames, filenames in os.walk(in_path):
-            relpath = os.path.relpath(dirpath, in_path)
-            for filename in dirnames + filenames:
-                callback('update', 1)
-                real_src = os.path.join(dirpath, filename)
-                real_dest = os.path.join(relpath, filename)
-                archive.write(real_src, real_dest)
-
-
-def compress_folder_tar(
-    in_path: Union[str, Path],
-    out_path: Union[str, Path],
-    *,
-    callback: Callable[[str, Any], None] = null_callback,
-    callback_description: str = 'Compressing objects as tar'
-):
-    """Compress a folder as a zip file
-
-    :param in_path: Path to compress
-    :param out_path: Path to compress to
-    :param callback: a callback to report on the process, ``callback(action, value)``,
-        with the following callback signatures:
-
-        - ``callback('init', {'total': <int>, 'description': <str>})``,
-            to signal the start of a process, its total iterations and description
-        - ``callback('update', <int>)``,
-            to signal an update to the process and the number of iterations to progress
-
-    :param callback_description: the description to return in the callback
-
-    """
-    callback('init', {'total': 1, 'description': 'Computing objects to compress'})
-    count = 0
-    for _, dirnames, filenames in os.walk(in_path):
-        count += len(dirnames) + len(filenames)
-    callback('init', {'total': count + 1, 'description': callback_description})
-
-    def _filter(tarinfo):
-        callback('update', 1)
-        return tarinfo
-
-    with tarfile.open(os.path.abspath(out_path), 'w:gz', format=tarfile.PAX_FORMAT, dereference=True) as archive:
-        archive.add(os.path.abspath(in_path), arcname='', filter=_filter)
-
-
 class CacheFolder:
     """A class to encapsulate a folder path with cached read/writes.
 
diff --git a/aiida/tools/importexport/archive/migrations/v03_to_v04.py b/aiida/tools/importexport/archive/migrations/v03_to_v04.py
index 55a3577484..b0c3fc97df 100644
--- a/aiida/tools/importexport/archive/migrations/v03_to_v04.py
+++ b/aiida/tools/importexport/archive/migrations/v03_to_v04.py
@@ -421,8 +421,8 @@ def add_extras(data):
     Since Extras were not available previously and usually only include hashes,
     the Node ids will be added, but included as empty dicts
     """
-    node_extras = {}
-    node_extras_conversion = {}
+    node_extras: dict = {}
+    node_extras_conversion: dict = {}
 
     for node_id in data['export_data'].get('Node', {}):
         node_extras[node_id] = {}
diff --git a/aiida/tools/importexport/archive/migrations/v05_to_v06.py b/aiida/tools/importexport/archive/migrations/v05_to_v06.py
index 8f2ff472fe..b0be661591 100644
--- a/aiida/tools/importexport/archive/migrations/v05_to_v06.py
+++ b/aiida/tools/importexport/archive/migrations/v05_to_v06.py
@@ -24,6 +24,8 @@
 Where id is a SQLA id and migration-name is the name of the particular migration.
 """
 # pylint: disable=invalid-name
+from typing import Union
+
 from aiida.tools.importexport.archive.common import CacheFolder
 
 from .utils import verify_metadata_version, update_metadata
@@ -33,6 +35,8 @@ def migrate_deserialized_datetime(data, conversion):
     """Deserialize datetime strings from export archives, meaning to reattach the UTC timezone information."""
     from aiida.tools.importexport.common.exceptions import ArchiveMigrationError
 
+    ret_data: Union[str, dict, list]
+
     if isinstance(data, dict):
         ret_data = {}
         for key, value in data.items():
diff --git a/aiida/tools/importexport/archive/migrators.py b/aiida/tools/importexport/archive/migrators.py
index 15cc1def3d..1446cc7ef2 100644
--- a/aiida/tools/importexport/archive/migrators.py
+++ b/aiida/tools/importexport/archive/migrators.py
@@ -13,18 +13,18 @@
 import os
 from pathlib import Path
 import shutil
+import tarfile
 import tempfile
 from typing import Any, Callable, cast, List, Optional, Type, Union
 import zipfile
 
+from archive_path import TarPath, ZipPath, read_file_in_tar, read_file_in_zip
+
 from aiida.common.log import AIIDA_LOGGER
 from aiida.common.progress_reporter import get_progress_reporter, create_callback
 from aiida.tools.importexport.common.exceptions import (ArchiveMigrationError, CorruptArchive, DanglingLinkError)
 from aiida.tools.importexport.common.config import ExportFileFormat
-from aiida.tools.importexport.archive.common import (
-    read_file_in_tar, read_file_in_zip, safe_extract_tar, safe_extract_zip, compress_folder_tar, compress_folder_zip,
-    CacheFolder
-)
+from aiida.tools.importexport.archive.common import CacheFolder
 from aiida.tools.importexport.archive.migrations import MIGRATE_FUNCTIONS
 
 __all__ = (
@@ -227,39 +227,54 @@ def _extract_archive(self, filepath: Path, callback: Callable[[str, Any], None])
     @staticmethod
     def _compress_archive_zip(in_path: Path, out_path: Path, compression: int):
         """Create a new zip compressed zip from a folder."""
-        with get_progress_reporter()(total=1) as progress:
+        with get_progress_reporter()(total=1, desc='Compressing to zip') as progress:
             _callback = create_callback(progress)
-            compress_folder_zip(in_path, out_path, compression=compression, callback=_callback)
+            with ZipPath(out_path, mode='w', compression=compression, allow_zip64=True) as path:
+                path.puttree(in_path, check_exists=False, callback=_callback, cb_descript='Compressing to zip')
 
     @staticmethod
     def _compress_archive_tar(in_path: Path, out_path: Path):
         """Create a new zip compressed tar from a folder."""
-        with get_progress_reporter()(total=1) as progress:
+        with get_progress_reporter()(total=1, desc='Compressing to tar') as progress:
             _callback = create_callback(progress)
-            compress_folder_tar(in_path, out_path, callback=_callback)
+            with TarPath(out_path, mode='w:gz', dereference=True) as path:
+                path.puttree(in_path, check_exists=False, callback=_callback, cb_descript='Compressing to tar')
 
 
 class ArchiveMigratorJsonZip(ArchiveMigratorJsonBase):
     """A migrator for a JSON zip compressed format."""
 
     def _retrieve_version(self) -> str:
-        metadata = json.loads(read_file_in_zip(self.filepath, 'metadata.json'))
+        try:
+            metadata = json.loads(read_file_in_zip(self.filepath, 'metadata.json'))
+        except (IOError, FileNotFoundError) as error:
+            raise CorruptArchive(str(error))
         if 'export_version' not in metadata:
             raise CorruptArchive("metadata.json doest not contain an 'export_version' key")
         return metadata['export_version']
 
     def _extract_archive(self, filepath: Path, callback: Callable[[str, Any], None]):
-        safe_extract_zip(self.filepath, filepath, callback=callback)
+        try:
+            ZipPath(self.filepath, mode='r', allow_zip64=True).extract_tree(filepath, callback=callback)
+        except zipfile.BadZipfile as error:
+            raise CorruptArchive(f'The input file cannot be read: {error}')
 
 
 class ArchiveMigratorJsonTar(ArchiveMigratorJsonBase):
     """A migrator for a JSON tar compressed format."""
 
     def _retrieve_version(self) -> str:
-        metadata = json.loads(read_file_in_tar(self.filepath, 'metadata.json'))
+        try:
+            metadata = json.loads(read_file_in_tar(self.filepath, 'metadata.json'))
+        except (IOError, FileNotFoundError) as error:
+            raise CorruptArchive(str(error))
         if 'export_version' not in metadata:
             raise CorruptArchive("metadata.json doest not contain an 'export_version' key")
         return metadata['export_version']
 
     def _extract_archive(self, filepath: Path, callback: Callable[[str, Any], None]):
-        safe_extract_tar(self.filepath, filepath, callback=callback)
+        try:
+            TarPath(self.filepath, mode='r:*', pax_format=tarfile.PAX_FORMAT
+                    ).extract_tree(filepath, allow_dev=False, allow_symlink=False, callback=callback)
+        except tarfile.ReadError as error:
+            raise CorruptArchive(f'The input file cannot be read: {error}')
diff --git a/aiida/tools/importexport/archive/readers.py b/aiida/tools/importexport/archive/readers.py
index 3fe7c46380..65da299ab2 100644
--- a/aiida/tools/importexport/archive/readers.py
+++ b/aiida/tools/importexport/archive/readers.py
@@ -12,19 +12,20 @@
 import json
 import os
 from pathlib import Path
+import tarfile
 from types import TracebackType
 from typing import Any, Callable, cast, Dict, Iterable, Iterator, List, Optional, Set, Tuple, Type
+import zipfile
 
 from distutils.version import StrictVersion
+from archive_path import TarPath, ZipPath, read_file_in_tar, read_file_in_zip
 
 from aiida.common.log import AIIDA_LOGGER
 from aiida.common.exceptions import InvalidOperation
 from aiida.common.folders import Folder, SandboxFolder
 from aiida.tools.importexport.common.config import EXPORT_VERSION, ExportFileFormat, NODES_EXPORT_SUBFOLDER
 from aiida.tools.importexport.common.exceptions import (CorruptArchive, IncompatibleArchiveVersionError)
-from aiida.tools.importexport.archive.common import (
-    ArchiveMetadata, null_callback, read_file_in_tar, read_file_in_zip, safe_extract_zip, safe_extract_tar
-)
+from aiida.tools.importexport.archive.common import (ArchiveMetadata, null_callback)
 from aiida.tools.importexport.common.config import NODE_ENTITY_NAME, GROUP_ENTITY_NAME
 from aiida.tools.importexport.common.utils import export_shard_uuid
 
@@ -367,7 +368,7 @@ def iter_node_repos(
         assert self._sandbox is not None  # required by mypy
 
         # unarchive the common folder if it does not exist
-        common_prefix = os.path.commonprefix(path_prefixes)
+        common_prefix = os.path.commonpath(path_prefixes)
         if not self._sandbox.get_subfolder(common_prefix).exists():
             self._extract(path_prefix=common_prefix, callback=callback)
 
@@ -391,24 +392,31 @@ def file_format_verbose(self) -> str:
 
     def _get_metadata(self):
         if self._metadata is None:
-            self._metadata = json.loads(read_file_in_zip(self.filename, self.FILENAME_METADATA))
+            try:
+                self._metadata = json.loads(read_file_in_zip(self.filename, self.FILENAME_METADATA))
+            except (IOError, FileNotFoundError) as error:
+                raise CorruptArchive(str(error))
         return self._metadata
 
     def _get_data(self):
         if self._data is None:
-            self._data = json.loads(read_file_in_zip(self.filename, self.FILENAME_DATA))
+            try:
+                self._data = json.loads(read_file_in_zip(self.filename, self.FILENAME_DATA))
+            except (IOError, FileNotFoundError) as error:
+                raise CorruptArchive(str(error))
         return self._data
 
     def _extract(self, *, path_prefix: str, callback: Callable[[str, Any], None] = null_callback):
         self.assert_within_context()
         assert self._sandbox is not None  # required by mypy
-        safe_extract_zip(
-            self.filename,
-            self._sandbox.abspath,
-            only_prefix=[path_prefix],
-            callback=callback,
-            callback_description='Extracting repository files'
-        )
+        try:
+            ZipPath(self.filename, mode='r', allow_zip64=True).joinpath(path_prefix).extract_tree(
+                self._sandbox.abspath, callback=callback, cb_descript='Extracting repository files'
+            )
+        except zipfile.BadZipfile as error:
+            raise CorruptArchive(f'The input file cannot be read: {error}')
+        except NotADirectoryError as error:
+            raise CorruptArchive(f'Unable to find required folder in archive: {error}')
 
 
 class ReaderJsonTar(ReaderJsonBase):
@@ -420,24 +428,35 @@ def file_format_verbose(self) -> str:
 
     def _get_metadata(self):
         if self._metadata is None:
-            self._metadata = json.loads(read_file_in_tar(self.filename, self.FILENAME_METADATA))
+            try:
+                self._metadata = json.loads(read_file_in_tar(self.filename, self.FILENAME_METADATA))
+            except (IOError, FileNotFoundError) as error:
+                raise CorruptArchive(str(error))
         return self._metadata
 
     def _get_data(self):
         if self._data is None:
-            self._data = json.loads(read_file_in_tar(self.filename, self.FILENAME_DATA))
+            try:
+                self._data = json.loads(read_file_in_tar(self.filename, self.FILENAME_DATA))
+            except (IOError, FileNotFoundError) as error:
+                raise CorruptArchive(str(error))
         return self._data
 
     def _extract(self, *, path_prefix: str, callback: Callable[[str, Any], None] = null_callback):
         self.assert_within_context()
         assert self._sandbox is not None  # required by mypy
-        safe_extract_tar(
-            self.filename,
-            self._sandbox.abspath,
-            only_prefix=[path_prefix],
-            callback=callback,
-            callback_description='Extracting repository files'
-        )
+        try:
+            TarPath(self.filename, mode='r:*').joinpath(path_prefix).extract_tree(
+                self._sandbox.abspath,
+                allow_dev=False,
+                allow_symlink=False,
+                callback=callback,
+                cb_descript='Extracting repository files'
+            )
+        except tarfile.ReadError as error:
+            raise CorruptArchive(f'The input file cannot be read: {error}')
+        except NotADirectoryError as error:
+            raise CorruptArchive(f'Unable to find required folder in archive: {error}')
 
 
 class ReaderJsonFolder(ReaderJsonBase):
diff --git a/aiida/tools/importexport/archive/writers.py b/aiida/tools/importexport/archive/writers.py
index 814795cc0a..a4d96ba7e2 100644
--- a/aiida/tools/importexport/archive/writers.py
+++ b/aiida/tools/importexport/archive/writers.py
@@ -9,20 +9,29 @@
 ###########################################################################
 """Archive writer classes."""
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
-import json
+from copy import deepcopy
+import os
+from pathlib import Path
+import shelve
+import shutil
 import time
-import tarfile
-from typing import Any, Callable, cast, Dict, Iterable, List, Set, Tuple, Type, Union
+import tempfile
+from types import TracebackType
+from typing import Any, cast, Dict, List, Optional, Type, Union
+import zipfile
 
-from aiida.common.folders import Folder, SandboxFolder
-from aiida.common.progress_reporter import get_progress_reporter
+from archive_path import TarPath, ZipPath
+
+from aiida.common import json
+from aiida.common.exceptions import InvalidOperation
+from aiida.common.folders import Folder
 from aiida.tools.importexport.archive.common import ArchiveMetadata
-from aiida.tools.importexport.common.config import EXPORT_VERSION, NODES_EXPORT_SUBFOLDER, ExportFileFormat
+from aiida.tools.importexport.common.config import (
+    EXPORT_VERSION, NODE_ENTITY_NAME, NODES_EXPORT_SUBFOLDER, ExportFileFormat
+)
 from aiida.tools.importexport.common.utils import export_shard_uuid
-from aiida.tools.importexport.common.zip_folder import ZipFolder
 
-__all__ = ('ArchiveData', 'ArchiveWriterAbstract', 'get_writer', 'WriterJsonFolder', 'WriterJsonTar', 'WriterJsonZip')
+__all__ = ('ArchiveWriterAbstract', 'get_writer', 'WriterJsonZip', 'WriterJsonTar', 'WriterJsonFolder')
 
 
 def get_writer(file_format: str) -> Type['ArchiveWriterAbstract']:
@@ -31,229 +40,380 @@ def get_writer(file_format: str) -> Type['ArchiveWriterAbstract']:
         ExportFileFormat.ZIP: WriterJsonZip,
         ExportFileFormat.TAR_GZIPPED: WriterJsonTar,
         'folder': WriterJsonFolder,
+        'null': WriterNull,
     }
 
     if file_format not in writers:
         raise ValueError(
-            f'Can only write in the formats: {tuple(writers.keys())}, please specify one for "file_format".'
+            f'Can only write in the formats: {tuple(writers.keys())}, not {file_format}, '
+            'please specify one for "file_format".'
         )
 
     return cast(Type[ArchiveWriterAbstract], writers[file_format])
 
 
-@dataclass
-class ArchiveData:
-    """Class for storing data, to export to an AiiDA archive."""
-    metadata: ArchiveMetadata
-    node_uuids: Set[str]
-    # UUID of the group -> UUIDs of the entities it contains
-    group_uuids: Dict[str, Set[str]]
-    # list of {'input': <UUID>, 'output': <UUID>, 'label': <LABEL>, 'type': <TYPE>}
-    link_uuids: List[dict]
-    # all entity data from the database, except Node extras and attributes
-    # {'ENTITY_NAME': {<Pk>: {'db_key': 'value', ...}, ...}, ...}
-    entity_data: Dict[str, Dict[int, dict]]
-    # Iterable of Node (pk, attributes, extras)
-    node_data: Iterable[Tuple[str, dict, dict]]
-
-    def __repr__(self) -> str:
-        """Return string representation."""
-        return f'ArchiveData(metadata={self.metadata},' + ','.join(
-            f'#{k}={len(v)}' for k, v in self.entity_data.items()
-        ) + ')'
-
-
 class ArchiveWriterAbstract(ABC):
     """An abstract interface for AiiDA archive writers."""
 
-    def __init__(self, filename: str, **kwargs: Any):
-        """An archive writer
+    def __init__(self, filepath: Union[str, Path], **kwargs: Any):
+        """Initiate the writer.
 
-        :param filename: the filename (possibly including the absolute path)
-            of the file on which to export.
+        :param filepath: the path to the file to export to.
+        :param kwargs: keyword arguments specific to the writer implementation.
 
         """
         # pylint: disable=unused-argument
-        self._filename = filename
+        self._filepath = Path(filepath)
+        self._info: Dict[str, Any] = {}
+        self._in_context: bool = False
+
+    @property
+    def filepath(self) -> Path:
+        """Return the filepath to write to."""
+        return self._filepath
 
     @property
-    def filename(self) -> str:
-        """Return the filename to write to."""
-        return self._filename
+    def export_info(self) -> Dict[str, Any]:
+        """Return information stored during the export process."""
+        return deepcopy(self._info)
 
     @property
     @abstractmethod
     def file_format_verbose(self) -> str:
-        """The file format name."""
+        """Return the file format name."""
 
     @property
     @abstractmethod
     def export_version(self) -> str:
-        """The export version."""
+        """Return the export version."""
+
+    def __enter__(self) -> 'ArchiveWriterAbstract':
+        """Open the contextmanager """
+        self._in_context = True
+        # reset the export information
+        self._info = {'writer_entered': time.time()}
+        self.open()
+        return self
+
+    def __exit__(
+        self, exctype: Optional[Type[BaseException]], excinst: Optional[BaseException], exctb: Optional[TracebackType]
+    ):
+        """Open the contextmanager """
+        self.close(exctype is not None)
+        self._in_context = False
+        self._info['writer_exited'] = time.time()
+        return False
+
+    def assert_within_context(self):
+        """Assert that the method is called within a context.
+
+        :raises: `~aiida.common.exceptions.InvalidOperation`: if not called within a context
+        """
+        if not self._in_context:
+            raise InvalidOperation('the ArchiveReader method should be used within a context')
+
+    def add_export_info(self, key: str, value: Any):
+        """Record information about the export process.
+
+        This information can be accessed by ``writer.export_info``,
+        it is reset on entrance of the context manager.
+
+        """
+        self._info[key] = value
+
+    def _remove_filepath(self):
+        """To run before moving the final export from a temporary folder."""
+        if self.filepath.exists():
+            if self.filepath.is_file():
+                self.filepath.unlink()
+            else:
+                shutil.rmtree(str(self.filepath))
+
+    # initialise/finalise methods
 
     @abstractmethod
-    def write(self, export_data: ArchiveData, get_node_repo: Callable[[str], Folder]) -> dict:
-        """write the archive.
+    def open(self):
+        """Setup the export process.
 
-        :param export_data: The data to export
-        :param get_node_repo: A callable: UUID -> to a folder of the node's repository
-        :returns: A dictionary of data about the write process
+        This method is called on entering a context manager.
 
         """
 
+    @abstractmethod
+    def close(self, excepted: bool):
+        """Finalise the export.
 
-def _write_to_json_archive(
-    folder: Union[Folder, ZipFolder], export_data: ArchiveData, export_version: str, get_node_repo: Callable[[str],
-                                                                                                             Folder]
-) -> None:
-    """Write data to the archive."""
-    # subfolder inside the export package
-    nodesubfolder = folder.get_subfolder(NODES_EXPORT_SUBFOLDER, create=True, reset_limit=True)
-
-    data: Dict[str, Any] = {
-        'node_attributes': {},
-        'node_extras': {},
-        'export_data': export_data.entity_data,
-        'links_uuid': export_data.link_uuids,
-        'groups_uuid': {key: list(vals) for key, vals in export_data.group_uuids.items()},
-    }
+        This method is called on exiting a context manager.
 
-    for uuid, attributes, extras in export_data.node_data:
-        data['node_attributes'][uuid] = attributes
-        data['node_extras'][uuid] = extras
-
-    # N.B. We're really calling zipfolder.open (if exporting a zipfile)
-    with folder.open('data.json', mode='w') as fhandle:
-        # fhandle.write(json.dumps(data, cls=UUIDEncoder))
-        fhandle.write(json.dumps(data))
-
-    metadata = {
-        'export_version': export_version,
-        'aiida_version': export_data.metadata.aiida_version,
-        'all_fields_info': export_data.metadata.all_fields_info,
-        'unique_identifiers': export_data.metadata.unique_identifiers,
-        'export_parameters': {
-            'graph_traversal_rules': export_data.metadata.graph_traversal_rules,
-            'entities_starting_set': export_data.metadata.entities_starting_set,
-            'include_comments': export_data.metadata.include_comments,
-            'include_logs': export_data.metadata.include_logs,
-        },
-        'conversion_info': export_data.metadata.conversion_info
-    }
+        :param excepted: Whether
 
-    with folder.open('metadata.json', 'w') as fhandle:
-        fhandle.write(json.dumps(metadata))
+        """
 
-    # If there are no nodes, there are no repository files to store
-    if not export_data.node_uuids:
-        return
+    # write methods
 
-    pbar_base_str = 'Exporting repository - '
-    with get_progress_reporter()(total=len(export_data.node_uuids)) as progress:
+    @abstractmethod
+    def write_metadata(self, data: ArchiveMetadata):
+        """Write the metadata of the export process."""
 
-        for uuid in export_data.node_uuids:
-            sharded_uuid = export_shard_uuid(uuid)
+    @abstractmethod
+    def write_link(self, data: Dict[str, str]):
+        """Write a dictionary of information for a single provenance link.
 
-            progress.set_description_str(f"{pbar_base_str}UUID={uuid.split('-')[0]}", refresh=False)
-            progress.update()
+        :param data: ``{'input': <UUID_STR>, 'output': <UUID_STR>, 'label': <LABEL_STR>, 'type': <TYPE_STR>}``
 
-            # Important to set create=False, otherwise creates twice a subfolder.
-            # Maybe this is a bug of insert_path?
-            thisnodefolder = nodesubfolder.get_subfolder(sharded_uuid, create=False, reset_limit=True)
-            src_folder = get_node_repo(uuid)
-            # In this way, I copy the content of the folder, and not the folder itself
-            thisnodefolder.insert_path(src=src_folder.abspath, dest_name='.')
+        """
 
+    @abstractmethod
+    def write_group_nodes(self, uuid: str, node_uuids: List[str]):
+        """Write a mapping of a group to the nodes it contains.
 
-class WriterJsonZip(ArchiveWriterAbstract):
-    """An archive writer,
-    which writes database data as a single JSON and repository data in a zipped folder system.
-    """
+        :param uuid: the UUID of the group
+        :param node_uuids: the list of node UUIDs the group contains
+
+        """
+
+    @abstractmethod
+    def write_entity_data(self, name: str, pk: int, id_key: str, fields: Dict[str, Any]):
+        """Write the data for a single DB entity.
+
+        :param name: the name of the entity (e.g. 'NODE')
+        :param pk: the primary key of the entity (unique for the current DB only)
+        :param id_key: the key within ``fields`` that denotes the unique identifier of the entity (unique for all DBs)
+        :param fields: mapping of DB fields to store -> values
 
-    def __init__(self, filename: str, use_compression: bool = True, **kwargs: Any):
-        """A writer for zipped archives.
+        """
+
+    @abstractmethod
+    def write_node_repo_folder(self, uuid: str, path: Union[str, Path], overwrite: bool = True):
+        """Write a node repository to the archive.
 
-        :param filename: the filename (possibly including the absolute path)
-            of the file on which to export.
-        :param use_compression: Whether or not to compress the zip file.
+        :param uuid: The UUID of the node
+        :param path: The path to the repository folder on disk
+        :param overwrite: Allow to overwrite existing path in archive
 
         """
-        super().__init__(filename, **kwargs)
-        self._use_compression = use_compression
+
+
+class WriterNull(ArchiveWriterAbstract):
+    """A null archive writer, which does not do anything."""
 
     @property
     def file_format_verbose(self) -> str:
-        return 'Zip (compressed)' if self._use_compression else 'Zip (uncompressed)'
+        return 'Null'
 
     @property
     def export_version(self) -> str:
         return EXPORT_VERSION
 
-    def write(self, export_data: ArchiveData, get_node_repo: Callable[[str], Folder]) -> dict:
-        """write the archive
+    def open(self):
+        pass
 
-        :param export_data: The data to export
-        :returns: A dictionary of data about the write process
+    def close(self, excepted: bool):
+        pass
 
-        """
-        with ZipFolder(self.filename, mode='w', use_compression=self._use_compression) as folder:
-            _write_to_json_archive(
-                folder=folder,
-                export_data=export_data,
-                export_version=self.export_version,
-                get_node_repo=get_node_repo,
-            )
+    def write_metadata(self, data: ArchiveMetadata):
+        pass
 
-        return {}
+    def write_link(self, data: Dict[str, str]):
+        pass
 
+    def write_group_nodes(self, uuid: str, node_uuids: List[str]):
+        pass
 
-class WriterJsonTar(ArchiveWriterAbstract):
-    """An archive writer,
-    which writes database data as a single JSON and repository data in a folder system.
+    def write_entity_data(self, name: str, pk: int, id_key: str, fields: Dict[str, Any]):
+        pass
 
-    The entire containing folder is then compressed as a tar file.
+    def write_node_repo_folder(self, uuid: str, path: Union[str, Path], overwrite: bool = True):
+        pass
+
+
+class WriterJsonZip(ArchiveWriterAbstract):
+    """An archive writer,
+    which writes database data as a single JSON and repository data in a zipped folder system.
     """
 
-    def __init__(self, filename: str, sandbox_in_repo: bool = True, **kwargs: Any):
-        """A writer for zipped archives.
+    def __init__(
+        self, filepath: Union[str, Path], *, use_compression: bool = True, cache_zipinfo: bool = False, **kwargs
+    ):
+        """Initiate the writer.
 
-        :param filename: the filename (possibly including the absolute path)
-            of the file on which to export.
-        :param sandbox_in_repo: Create the temporary uncompressed folder within the aiida repository
+        :param filepath: the path to the file to export to.
+        :param use_compression: Whether or not to deflate the objects inside the zip file.
+        :param cache_zipinfo: Cache the zip file index on disk during the write.
+            This reduces the RAM usage of the process, but will make the process slower.
 
         """
-        super().__init__(filename, **kwargs)
-        self.sandbox_in_repo = sandbox_in_repo
+        super().__init__(filepath)
+        self._compression = zipfile.ZIP_DEFLATED if use_compression else zipfile.ZIP_STORED
+        self._cache_zipinfo = cache_zipinfo
 
     @property
     def file_format_verbose(self) -> str:
-        return 'Gzipped tarball (compressed)'
+        return f'JSON Zip (compression={self._compression})'
 
     @property
     def export_version(self) -> str:
         return EXPORT_VERSION
 
-    def write(self, export_data: ArchiveData, get_node_repo: Callable[[str], Folder]) -> dict:
-        """write the archive
+    def open(self):
+        # pylint: disable=attribute-defined-outside-init
+        self.assert_within_context()
+        # create a temporary folder in which to perform the write
+        self._temp_path: Path = Path(tempfile.mkdtemp())
+        # open a zipfile in in write mode to export to
+        if self._cache_zipinfo:
+            self._zipinfo_cache = shelve.open(str(self._temp_path / 'zipinfo_cache'))
+        else:
+            self._zipinfo_cache = None
+        self._archivepath: ZipPath = ZipPath(
+            self._temp_path / 'export', mode='w', compression=self._compression, name_to_info=self._zipinfo_cache
+        )
+        # setup data to store
+        self._data: Dict[str, Any] = {
+            'node_attributes': {},
+            'node_extras': {},
+            'export_data': {},
+            'links_uuid': [],
+            'groups_uuid': {},
+        }
+
+    def close(self, excepted: bool):
+        self.assert_within_context()
+        if excepted:
+            self._archivepath.close()
+            shutil.rmtree(self._temp_path)
+            return
+        # write data.json
+        with self._archivepath.joinpath('data.json').open('wb') as handle:
+            json.dump(self._data, handle)
+        # close the zipfile to finalise write
+        self._archivepath.close()
+        if getattr(self, '_zipinfo_cache', None) is not None:
+            self._zipinfo_cache.close()
+            delattr(self, '_zipinfo_cache')
+        # move the compressed file to the final path
+        self._remove_filepath()
+        shutil.move(str(self._archivepath.filepath), str(self.filepath))
+        # remove the temporary folder
+        shutil.rmtree(self._temp_path)
+
+    def write_metadata(self, data: ArchiveMetadata):
+        metadata = {
+            'export_version': self.export_version,
+            'aiida_version': data.aiida_version,
+            'all_fields_info': data.all_fields_info,
+            'unique_identifiers': data.unique_identifiers,
+            'export_parameters': {
+                'graph_traversal_rules': data.graph_traversal_rules,
+                'entities_starting_set': data.entities_starting_set,
+                'include_comments': data.include_comments,
+                'include_logs': data.include_logs,
+            },
+            'conversion_info': data.conversion_info
+        }
+        with self._archivepath.joinpath('metadata.json').open('wb') as handle:
+            json.dump(metadata, handle)
+
+    def write_link(self, data: Dict[str, str]):
+        self._data['links_uuid'].append(data)
+
+    def write_group_nodes(self, uuid: str, node_uuids: List[str]):
+        self._data['groups_uuid'][uuid] = node_uuids
+
+    def write_entity_data(self, name: str, pk: int, id_key: str, fields: Dict[str, Any]):
+        if name == NODE_ENTITY_NAME:
+            # perform translation to current internal format
+            self._data['node_attributes'][pk] = fields.pop('attributes')
+            self._data['node_extras'][pk] = fields.pop('extras')
+        self._data['export_data'].setdefault(name, {})[pk] = fields
+
+    def write_node_repo_folder(self, uuid: str, path: Union[str, Path], overwrite: bool = True):
+        self.assert_within_context()
+        (self._archivepath / NODES_EXPORT_SUBFOLDER / export_shard_uuid(uuid)).puttree(path, check_exists=not overwrite)
 
-        :param export_data: The data to export
-        :returns: A dictionary of data about the write process
 
-        """
-        with SandboxFolder(sandbox_in_repo=self.sandbox_in_repo) as folder:
-            _write_to_json_archive(
-                folder=folder,
-                export_data=export_data,
-                export_version=self.export_version,
-                get_node_repo=get_node_repo,
-            )
+class WriterJsonTar(ArchiveWriterAbstract):
+    """An archive writer,
+    which writes database data as a single JSON and repository data in a folder system.
+
+    The entire containing folder is then compressed as a tar file.
+    """
+
+    @property
+    def file_format_verbose(self) -> str:
+        return 'Gzipped tarball (compressed)'
 
-            with tarfile.open(self.filename, 'w:gz', format=tarfile.PAX_FORMAT, dereference=True) as tar:
-                time_compress_start = time.time()
-                tar.add(folder.abspath, arcname='')
-                time_compress_stop = time.time()
+    @property
+    def export_version(self) -> str:
+        return EXPORT_VERSION
 
-        return {'compression_time_start': time_compress_start, 'compression_time_stop': time_compress_stop}
+    def open(self):
+        # pylint: disable=attribute-defined-outside-init
+        self.assert_within_context()
+        # create a temporary folder in which to perform the write
+        self._temp_path: Path = Path(tempfile.mkdtemp())
+        # open a zipfile in in write mode to export to
+        self._archivepath: TarPath = TarPath(self._temp_path / 'export', mode='w:gz', dereference=True)
+        # setup data to store
+        self._data: Dict[str, Any] = {
+            'node_attributes': {},
+            'node_extras': {},
+            'export_data': {},
+            'links_uuid': [],
+            'groups_uuid': {},
+        }
+
+    def close(self, excepted: bool):
+        self.assert_within_context()
+        if excepted:
+            self._archivepath.close()
+            shutil.rmtree(self._temp_path)
+            return
+        # write data.json
+        with self._archivepath.joinpath('data.json').open('wb') as handle:
+            json.dump(self._data, handle)
+        # compress
+        # close the zipfile to finalise write
+        self._archivepath.close()
+        # move the compressed file to the final path
+        self._remove_filepath()
+        shutil.move(str(self._archivepath.filepath), str(self.filepath))
+        # remove the temporary folder
+        shutil.rmtree(self._temp_path)
+
+    def write_metadata(self, data: ArchiveMetadata):
+        metadata = {
+            'export_version': self.export_version,
+            'aiida_version': data.aiida_version,
+            'all_fields_info': data.all_fields_info,
+            'unique_identifiers': data.unique_identifiers,
+            'export_parameters': {
+                'graph_traversal_rules': data.graph_traversal_rules,
+                'entities_starting_set': data.entities_starting_set,
+                'include_comments': data.include_comments,
+                'include_logs': data.include_logs,
+            },
+            'conversion_info': data.conversion_info
+        }
+        with self._archivepath.joinpath('metadata.json').open('wb') as handle:
+            json.dump(metadata, handle)
+
+    def write_link(self, data: Dict[str, str]):
+        self._data['links_uuid'].append(data)
+
+    def write_group_nodes(self, uuid: str, node_uuids: List[str]):
+        self._data['groups_uuid'][uuid] = node_uuids
+
+    def write_entity_data(self, name: str, pk: int, id_key: str, fields: Dict[str, Any]):
+        if name == NODE_ENTITY_NAME:
+            # perform translation to current internal format
+            self._data['node_attributes'][pk] = fields.pop('attributes')
+            self._data['node_extras'][pk] = fields.pop('extras')
+        self._data['export_data'].setdefault(name, {})[pk] = fields
+
+    def write_node_repo_folder(self, uuid: str, path: Union[str, Path], overwrite: bool = True):
+        self.assert_within_context()
+        (self._archivepath / NODES_EXPORT_SUBFOLDER / export_shard_uuid(uuid)).puttree(path, check_exists=not overwrite)
 
 
 class WriterJsonFolder(ArchiveWriterAbstract):
@@ -263,39 +423,78 @@ class WriterJsonFolder(ArchiveWriterAbstract):
     This writer is mainly intended for backward compatibility with `export_tree`.
     """
 
-    def __init__(self, filename: str, folder: Union[Folder, ZipFolder] = None, **kwargs: Any):
-        """A writer for zipped archives.
+    def __init__(self, filepath: str, folder: Folder = None, **kwargs: Any):
+        """Initiate the writer.
 
-        :param filename: the filename (possibly including the absolute path)
-            of the file on which to export.
         :param folder: a folder to write the archive to.
+        :param filepath: dummy value not used
 
         """
-        super().__init__(filename, **kwargs)
-        if not isinstance(folder, (Folder, ZipFolder)):
+        super().__init__(filepath, **kwargs)
+        if not isinstance(folder, Folder):
             raise TypeError('`folder` must be specified and given as an AiiDA Folder entity')
-        self._folder = cast(Union[Folder, ZipFolder], folder)
+        self._folder = cast(Folder, folder)
 
     @property
     def file_format_verbose(self) -> str:
-        return str(self._folder.__class__)
+        return 'JSON Folder'
 
     @property
     def export_version(self) -> str:
         return EXPORT_VERSION
 
-    def write(self, export_data: ArchiveData, get_node_repo: Callable[[str], Folder]) -> dict:
-        """write the archive
-
-        :param export_data: The data to export
-        :returns: A dictionary of data about the write process
-
-        """
-        _write_to_json_archive(
-            folder=self._folder,
-            export_data=export_data,
-            export_version=self.export_version,
-            get_node_repo=get_node_repo,
-        )
-
-        return {}
+    def open(self):
+        # pylint: disable=attribute-defined-outside-init
+        self.assert_within_context()
+        # setup data to store
+        self._data: Dict[str, Any] = {
+            'node_attributes': {},
+            'node_extras': {},
+            'export_data': {},
+            'links_uuid': [],
+            'groups_uuid': {},
+        }
+        # ensure folder is created
+        self._folder.create()
+
+    def close(self, excepted: bool):
+        self.assert_within_context()
+        if excepted:
+            return
+        with self._folder.open('data.json', 'wb') as handle:
+            json.dump(self._data, handle)
+
+    def write_metadata(self, data: ArchiveMetadata):
+        metadata = {
+            'export_version': self.export_version,
+            'aiida_version': data.aiida_version,
+            'all_fields_info': data.all_fields_info,
+            'unique_identifiers': data.unique_identifiers,
+            'export_parameters': {
+                'graph_traversal_rules': data.graph_traversal_rules,
+                'entities_starting_set': data.entities_starting_set,
+                'include_comments': data.include_comments,
+                'include_logs': data.include_logs,
+            },
+            'conversion_info': data.conversion_info
+        }
+        with self._folder.open('metadata.json', 'wb') as handle:
+            json.dump(metadata, handle)
+
+    def write_link(self, data: Dict[str, str]):
+        self._data['links_uuid'].append(data)
+
+    def write_group_nodes(self, uuid: str, node_uuids: List[str]):
+        self._data['groups_uuid'][uuid] = node_uuids
+
+    def write_entity_data(self, name: str, pk: int, id_key: str, fields: Dict[str, Any]):
+        if name == NODE_ENTITY_NAME:
+            # perform translation to current internal format
+            self._data['node_attributes'][pk] = fields.pop('attributes')
+            self._data['node_extras'][pk] = fields.pop('extras')
+        self._data['export_data'].setdefault(name, {})[pk] = fields
+
+    def write_node_repo_folder(self, uuid: str, path: Union[str, Path], overwrite: bool = True):
+        self.assert_within_context()
+        repo_folder = self._folder.get_subfolder(NODES_EXPORT_SUBFOLDER).get_subfolder(export_shard_uuid(uuid))
+        repo_folder.insert_path(src=os.path.abspath(path), dest_name='.', overwrite=overwrite)
diff --git a/aiida/tools/importexport/common/archive.py b/aiida/tools/importexport/common/archive.py
index c698348505..203b4f93b7 100644
--- a/aiida/tools/importexport/common/archive.py
+++ b/aiida/tools/importexport/common/archive.py
@@ -61,8 +61,8 @@ def extract_zip(infile, folder, nodes_export_subfolder=None, check_files=('data.
         incorrect formats
     """
     warnings.warn(
-        'extract_zip function is deprecated and will be removed in AiiDA v2.0.0, use `safe_extract_zip` instead',
-        AiidaDeprecationWarning
+        'extract_zip function is deprecated and will be removed in AiiDA v2.0.0, '
+        'use extract_tree in the archive-path package instead', AiidaDeprecationWarning
     )  # pylint: disable=no-member
 
     if nodes_export_subfolder:
@@ -135,8 +135,8 @@ def extract_tar(infile, folder, nodes_export_subfolder=None, check_files=('data.
         incorrect formats
     """
     warnings.warn(
-        'extract_tar function is deprecated and will be removed in AiiDA v2.0.0, use `safe_extract_tar` instead',
-        AiidaDeprecationWarning
+        'extract_tar function is deprecated and will be removed in AiiDA v2.0.0, '
+        'use extract_tree in the archive-path package instead', AiidaDeprecationWarning
     )  # pylint: disable=no-member
 
     if nodes_export_subfolder:
diff --git a/aiida/tools/importexport/common/config.py b/aiida/tools/importexport/common/config.py
index c92ae3084b..d7c52a79ea 100644
--- a/aiida/tools/importexport/common/config.py
+++ b/aiida/tools/importexport/common/config.py
@@ -178,7 +178,13 @@ def get_all_fields_info():
             'related_name': 'dbnodes'
         },
         'description': {},
-        'process_type': {}
+        'process_type': {},
+        'extras': {
+            'convert_type': 'jsonb'
+        },
+        'attributes': {
+            'convert_type': 'jsonb'
+        }
     }
     all_fields_info[GROUP_ENTITY_NAME] = {
         'description': {},
diff --git a/aiida/tools/importexport/common/zip_folder.py b/aiida/tools/importexport/common/zip_folder.py
deleted file mode 100644
index 9fc08d95e1..0000000000
--- a/aiida/tools/importexport/common/zip_folder.py
+++ /dev/null
@@ -1,196 +0,0 @@
-# -*- coding: utf-8 -*-
-###########################################################################
-# Copyright (c), The AiiDA team. All rights reserved.                     #
-# This file is part of the AiiDA code.                                    #
-#                                                                         #
-# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
-# For further information on the license, see the LICENSE.txt file        #
-# For further information please visit http://www.aiida.net               #
-###########################################################################
-"""Convenience class for interacting with a zipped folder."""
-import io
-import os
-from pathlib import Path
-from types import TracebackType
-from typing import IO, Optional, Type, Union
-import zipfile
-
-
-class _ZipFileWriter:
-    """An open zip file."""
-
-    def __init__(self, zip_file: zipfile.ZipFile, fname: str):
-        self._zipfile = zip_file
-        self._fname = fname
-        self._buffer: Optional[io.StringIO] = None
-
-    def open(self):
-        """Open the file for writing."""
-        if self._buffer is not None:
-            raise IOError('Cannot open again!')
-        self._buffer = io.StringIO()
-
-    def write(self, data: str):
-        """Write to the file."""
-        if self._buffer is None:
-            raise IOError('The file writer is not open')
-        self._buffer.write(data)
-
-    def close(self):
-        """Close to the file."""
-        self._buffer.seek(0)
-        self._zipfile.writestr(self._fname, self._buffer.read())
-        self._buffer = None
-
-    def __enter__(self):
-        """Enter the file for writing."""
-        self.open()
-        return self
-
-    def __exit__(
-        self, exctype: Optional[Type[BaseException]], excinst: Optional[BaseException], exctb: Optional[TracebackType]
-    ):
-        """Close the file."""
-        self.close()
-
-
-class ZipFolder:
-    """A wrapper around ``zipfile.ZipFile``, to provide convenience methods for interacting with a zipped folder."""
-
-    # To improve: if zipfile is closed, do something
-    # (e.g. add explicit open method, rename open to openfile, set _zipfile to None, ...)
-
-    def __init__(
-        self,
-        zipfolder_or_fname: Union[str, 'ZipFolder'],
-        mode: Optional[str] = None,
-        subfolder: str = '.',
-        use_compression: bool = True,
-        allowZip64: bool = True
-    ):
-        """Initialise the zip folder.
-
-        :param zipfolder_or_fname: either another ZipFolder instance,
-          of which you want to get a subfolder, or a filename to create.
-
-        :param mode: the file mode; see the zipfile.ZipFile docs for valid strings.
-            Note: can be specified only if zipfolder_or_fname is a string (the filename to generate)
-
-        :param subfolder: the subfolder that specified the "current working
-          directory" in the zip file. If zipfolder_or_fname is a ZipFolder,
-          subfolder is a relative path from zipfolder_or_fname.subfolder
-
-        :param use_compression: either True, to compress files in the Zip, or
-          False if you just want to pack them together without compressing.
-          It is ignored if zipfolder_or_fname is a ZipFolder instance.
-
-        """
-        if isinstance(zipfolder_or_fname, str):
-            the_mode = mode
-            if the_mode is None:
-                the_mode = 'r'
-            if use_compression:
-                compression = zipfile.ZIP_DEFLATED
-            else:
-                compression = zipfile.ZIP_STORED
-            self._zipfile = zipfile.ZipFile(
-                zipfolder_or_fname, mode=the_mode, compression=compression, allowZip64=allowZip64
-            )
-            self._pwd = subfolder
-        else:
-            if mode is not None:
-                raise ValueError("Cannot specify 'mode' when passing a ZipFolder")
-            self._zipfile = zipfolder_or_fname._zipfile  # pylint: disable=protected-access
-            self._pwd = os.path.join(zipfolder_or_fname.pwd, subfolder)
-
-    def __enter__(self):
-        """Enter the zipfile for reading/writing."""
-        return self
-
-    def __exit__(
-        self, exctype: Optional[Type[BaseException]], excinst: Optional[BaseException], exctb: Optional[TracebackType]
-    ):
-        """Exit the zipfile and close."""
-        self.close()
-
-    def close(self):
-        """Close the zipfile."""
-        self._zipfile.close()
-
-    @property
-    def pwd(self) -> str:
-        """Return the pathname working directory within the zipfile."""
-        return self._pwd
-
-    def open(self, fname: str, mode: str = 'r') -> Union['_ZipFileWriter', IO]:
-        """Open a path in the zipfile, for reading/writing
-
-        :param fname: An internal path in the zipfile
-        :param mode: mode should be 'r' to read a file already in the ZIP file,
-            or 'w' to write to a file newly added to the archive.
-
-        """
-        if mode == 'w':
-            return _ZipFileWriter(zip_file=self._zipfile, fname=self._get_internal_path(fname))
-        return self._zipfile.open(self._get_internal_path(fname), mode=mode)
-
-    def _get_internal_path(self, filename: str):
-        """Return a path to the filename from the root of zip file."""
-        return os.path.normpath(os.path.join(self.pwd, filename))
-
-    # pylint: disable=unused-argument
-    def get_subfolder(self, subfolder: str, create: bool = False, reset_limit: bool = False) -> 'ZipFolder':
-        """Return a subfolder in the zipfile as a new zip folder."""
-        return ZipFolder(self, subfolder=subfolder)
-
-    def exists(self, path):
-        """Check whether path already exists in the ZipFolder"""
-        try:
-            self._zipfile.getinfo(path)
-            exists = True
-        except KeyError:
-            exists = False
-        return exists
-
-    def insert_path(self, src: Union[str, Path], dest_name: Optional[Union[str, Path]] = None, overwrite: bool = True):
-        """Copy an external path into the zipfile
-
-        :param src: the path to the file or folder
-        :param dest_name: the path to write to in the zip file
-            If None, use the basename of src
-        :param overwrite: Overwrite any existing path in the zipfile
-
-        :raises ValueError: If the src path is not absolute
-        :raises IOError: If the destination already exists and overwrite is False
-        """
-        if dest_name is None:
-            base_filename = str(os.path.basename(src))
-        else:
-            base_filename = str(dest_name)
-
-        base_filename = self._get_internal_path(base_filename)
-
-        src = str(src)
-
-        if not os.path.isabs(src):
-            raise ValueError('src must be an absolute path in insert_file')
-
-        if not overwrite and self.exists(base_filename):
-            raise IOError(f'destination already exists: {base_filename}')
-
-        if os.path.isdir(src):
-            for dirpath, dirnames, filenames in os.walk(src):
-                relpath = os.path.relpath(dirpath, src)
-                if not dirnames and not filenames:
-                    real_src = dirpath
-                    real_dest = os.path.join(base_filename, relpath) + os.path.sep
-                    if not self.exists(real_dest):
-                        self._zipfile.write(real_src, real_dest)
-                else:
-                    for fname in dirnames + filenames:
-                        real_src = os.path.join(dirpath, fname)
-                        real_dest = os.path.join(base_filename, relpath, fname)
-                        if not self.exists(real_dest):
-                            self._zipfile.write(real_src, real_dest)
-        else:
-            self._zipfile.write(src, base_filename)
diff --git a/aiida/tools/importexport/dbexport/__init__.py b/aiida/tools/importexport/dbexport/__init__.py
index 652a00ae30..3476706c32 100644
--- a/aiida/tools/importexport/dbexport/__init__.py
+++ b/aiida/tools/importexport/dbexport/__init__.py
@@ -11,7 +11,6 @@
 """Provides export functionalities."""
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from dataclasses import dataclass, field, asdict
 import logging
 import os
 import tarfile
@@ -60,7 +59,7 @@
     model_fields_to_file_fields,
 )
 from aiida.tools.graph.graph_traversers import get_nodes_export, validate_traversal_rules
-from aiida.tools.importexport.archive.writers import ArchiveData, ArchiveMetadata, get_writer
+from aiida.tools.importexport.archive.writers import ArchiveMetadata, ArchiveWriterAbstract, get_writer
 from aiida.tools.importexport.common.config import ExportFileFormat
 from aiida.tools.importexport.common.utils import export_shard_uuid
 from aiida.tools.importexport.dbexport.utils import (
@@ -72,43 +71,25 @@
     serialize_dict,
     summary,
 )
-from aiida.tools.importexport.common.zip_folder import ZipFolder
 
 __all__ = ('export', 'EXPORT_LOGGER', 'ExportFileFormat')
 
 
-@dataclass
-class ExportReport:
-    """Class for storing data about the export process."""
-    # time in seconds
-    time_collect_start: float
-    time_collect_stop: float
-    # skipped if no data to write
-    time_write_start: Optional[float] = None
-    time_write_stop: Optional[float] = None
-    # additional data returned  by the writer
-    writer_data: Optional[Dict[str, Any]] = None
-
-    @property
-    def total_time(self) -> float:
-        """Return total time taken in seconds."""
-        return (self.time_write_stop or self.time_collect_stop) - self.time_collect_start
-
-
 def export(
     entities: Optional[Iterable[Any]] = None,
     filename: Optional[str] = None,
-    file_format: str = ExportFileFormat.ZIP,
+    file_format: Union[str, Type[ArchiveWriterAbstract]] = ExportFileFormat.ZIP,
     overwrite: bool = False,
     silent: Optional[bool] = None,
-    use_compression: bool = True,
+    use_compression: Optional[bool] = None,
     include_comments: bool = True,
     include_logs: bool = True,
     allowed_licenses: Optional[Union[list, Callable]] = None,
     forbidden_licenses: Optional[Union[list, Callable]] = None,
     writer_init: Optional[Dict[str, Any]] = None,
+    batch_size: int = 100,
     **traversal_rules: bool,
-) -> ExportReport:
+) -> ArchiveWriterAbstract:
     """Export AiiDA data to an archive file.
 
     Note, the logging level and progress reporter should be set externally, for example::
@@ -123,6 +104,10 @@ def export(
         Support for the parameter `silent` will be removed in `v2.0.0`.
         Please set the log level and progress bar implementation independently.
 
+    .. deprecated:: 1.5.0
+        Support for the parameter `use_compression` will be removed in `v2.0.0`.
+        Please use `writer_init={'use_compression': True}`.
+
     .. deprecated:: 1.2.1
         Support for the parameters `what` and `outfile` will be removed in `v2.0.0`.
         Please use `entities` and `filename` instead, respectively.
@@ -133,16 +118,13 @@ def export(
     :param filename: the filename (possibly including the absolute path)
         of the file on which to export.
 
-    :param file_format: See `ExportFileFormat` for complete list of valid values (default: 'zip').
+    :param file_format: 'zip', 'tar.gz' or 'folder' or a specific writer class.
 
     :param overwrite: if True, overwrite the output file without asking, if it exists.
         If False, raise an
         :py:class:`~aiida.tools.importexport.common.exceptions.ArchiveExportError`
         if the output file already exists.
 
-    :param use_compression: Whether or not to compress the archive file
-        (only valid for the zip file format).
-
     :param allowed_licenses: List or function.
         If a list, then checks whether all licenses of Data nodes are in the list. If a function,
         then calls function for licenses of Data nodes expecting True if license is allowed, False
@@ -161,6 +143,8 @@ def export(
 
     :param writer_init: Additional key-word arguments to pass to the writer class init
 
+    :param batch_size: batch database query results in sub-collections to reduce memory usage
+
     :param traversal_rules: graph traversal rules. See :const:`aiida.common.links.GraphTraversalRules`
         what rule names are toggleable and what the defaults are.
 
@@ -171,6 +155,8 @@ def export(
     :raises `~aiida.common.exceptions.LicensingException`:
         if any node is licensed under forbidden license.
     """
+    # pylint: disable=too-many-locals,too-many-branches,too-many-statements
+
     # Backwards-compatibility
     entities = cast(
         Iterable[Any],
@@ -222,7 +208,20 @@ def export(
         name: traversal_rules.get(name, rule.default) for name, rule in GraphTraversalRules.EXPORT.value.items()
     }
 
-    writer = get_writer(file_format)(filename=filename, use_compression=use_compression, **(writer_init or {}))
+    # setup the archive writer
+    writer_init = writer_init or {}
+    if use_compression is not None:
+        warnings.warn(
+            'use_compression argument is deprecated and will be removed in AiiDA v2.0.0 (which will always compress)',
+            AiidaDeprecationWarning
+        )  # pylint: disable=no-member
+        writer_init['use_compression'] = use_compression
+    if isinstance(file_format, str):
+        writer = get_writer(file_format)(filepath=filename, **writer_init)
+    elif issubclass(file_format, ArchiveWriterAbstract):
+        writer = file_format(filepath=filename, **writer_init)
+    else:
+        raise TypeError('file_format must be a string or ArchiveWriterAbstract class')
 
     summary(
         file_format=writer.file_format_verbose,
@@ -233,193 +232,116 @@ def export(
         traversal_rules=full_traversal_rules
     )
 
-    report_data: Dict[str, Any] = {'time_write_start': None, 'time_write_stop': None, 'writer_data': None}
-
-    report_data['time_collect_start'] = time.time()
-    export_data = _collect_archive_data(
-        entities=entities,
-        allowed_licenses=allowed_licenses,
-        forbidden_licenses=forbidden_licenses,
-        include_comments=include_comments,
-        include_logs=include_logs,
-        **traversal_rules
-    )
-    report_data['time_collect_stop'] = time.time()
-
-    extract_time = report_data['time_collect_stop'] - report_data['time_collect_start']
-    EXPORT_LOGGER.debug(f'Data extracted in {extract_time:6.2g} s.')
-
-    if export_data is not None:
-
-        EXPORT_LOGGER.debug('ADDING DATA TO EXPORT ARCHIVE...')
-
-        def _get_node_repo(uuid):
-            # Make sure the node's repository folder was not deleted
-            src = RepositoryFolder(section=Repository._section_name, uuid=uuid)  # pylint: disable=protected-access
-            if not src.exists():
-                raise exceptions.ArchiveExportError(
-                    f'Unable to find the repository folder for Node with UUID={uuid} '
-                    'in the local repository'
-                )
-            return src
-
-        try:
-            report_data['time_write_start'] = time.time()
-            report_data['writer_data'] = writer.write(
-                export_data=export_data, get_node_repo=_get_node_repo
-            )  # type: ignore
-            report_data['time_write_stop'] = time.time()
-        except (exceptions.ArchiveExportError, LicensingException) as exc:
-            if os.path.exists(filename):
-                os.remove(filename)
-            raise exc
-
-        write_time = report_data['time_write_stop'] - report_data['time_write_start']
-        EXPORT_LOGGER.debug(f'Data written in {write_time:6.2g} s.')
-
-    else:
-        EXPORT_LOGGER.debug('No data to write.')
-
-    return ExportReport(**report_data)
-
-
-def _collect_archive_data(
-    entities: Optional[Iterable[Any]] = None,
-    allowed_licenses: Optional[Union[list, Callable]] = None,
-    forbidden_licenses: Optional[Union[list, Callable]] = None,
-    include_comments: bool = True,
-    include_logs: bool = True,
-    **traversal_rules: bool,
-) -> Optional[ArchiveData]:
-    """Collect data to be exported
-
-    .. deprecated:: 1.2.1
-        Support for the parameter `what` will be removed in `v2.0.0`. Please use `entities` instead.
-
-    :param entities: a list of entity instances; they can belong to different models/entities.
-
-    :param allowed_licenses: List or function.
-        If a list, then checks whether all licenses of Data nodes are in the list.
-        If a function, then calls function for licenses of Data nodes,
-        expecting True if license is allowed, False otherwise.
-
-    :param forbidden_licenses: List or function.
-        If a list, then checks whether all licenses of Data nodes are in the list.
-        If a function, then calls function for licenses of Data nodes,
-        expecting True if license is allowed, False otherwise.
-
-    :param include_comments: In-/exclude export of comments for given node(s) in ``entities``.
-        Default: True, *include* comments in export (as well as relevant users).
+    EXPORT_LOGGER.debug('STARTING EXPORT...')
 
-    :param include_logs: In-/exclude export of logs for given node(s) in ``entities``.
-        Default: True, *include* logs in export.
+    all_fields_info, unique_identifiers = get_all_fields_info()
+    entities_starting_set, given_node_entry_ids = _get_starting_node_ids(entities)
 
-    :param traversal_rules: graph traversal rules.
-        See :const:`aiida.common.links.GraphTraversalRules`
-        what rule names are toggleable and what the defaults are.
+    # Initialize the writer
+    with writer as writer_context:
 
-    :raises `~aiida.tools.importexport.common.exceptions.ArchiveExportError`:
-        if there are any internal errors when exporting.
-    :raises `~aiida.common.exceptions.LicensingException`:
-        if any node is licensed under forbidden license.
-    """
-    # pylint: disable=too-many-locals
-    EXPORT_LOGGER.debug('STARTING EXPORT...')
+        # Iteratively explore the AiiDA graph to find further nodes that should also be exported
+        with get_progress_reporter()(desc='Traversing provenance via links ...', total=1) as progress:
+            traverse_output = get_nodes_export(starting_pks=given_node_entry_ids, get_links=True, **traversal_rules)
+            progress.update()
+        node_ids_to_be_exported = traverse_output['nodes']
+
+        EXPORT_LOGGER.debug('WRITING METADATA...')
+
+        writer_context.write_metadata(
+            ArchiveMetadata(
+                export_version=EXPORT_VERSION,
+                aiida_version=get_version(),
+                unique_identifiers=unique_identifiers,
+                all_fields_info=all_fields_info,
+                graph_traversal_rules=traverse_output['rules'],
+                # Turn sets into lists to be able to export them as JSON metadata.
+                entities_starting_set={
+                    entity: list(entity_set) for entity, entity_set in entities_starting_set.items()
+                },
+                include_comments=include_comments,
+                include_logs=include_logs,
+            )
+        )
 
-    # Backwards-compatibility
-    entities = cast(
-        Iterable[Any],
-        deprecated_parameters(
-            old={
-                'name': 'what',
-                'value': traversal_rules.pop('what', None)
-            },
-            new={
-                'name': 'entities',
-                'value': entities
-            },
-        ),
-    )
+        # Create a mapping of node PK to UUID.
+        node_pk_2_uuid_mapping: Dict[int, str] = {}
+        if node_ids_to_be_exported:
+            qbuilder = orm.QueryBuilder().append(
+                orm.Node,
+                project=('id', 'uuid'),
+                filters={'id': {
+                    'in': node_ids_to_be_exported
+                }},
+            )
+            node_pk_2_uuid_mapping = dict(qbuilder.all(batch_size=batch_size))
+
+        # check that no nodes are being exported with incorrect licensing
+        _check_node_licenses(node_ids_to_be_exported, allowed_licenses, forbidden_licenses)
+
+        # write the link data
+        if traverse_output['links']:
+            with get_progress_reporter()(total=len(traverse_output['links']), desc='Writing links') as progress:
+                for link in traverse_output['links']:
+                    progress.update()
+                    writer_context.write_link({
+                        'input': node_pk_2_uuid_mapping[link.source_id],
+                        'output': node_pk_2_uuid_mapping[link.target_id],
+                        'label': link.link_label,
+                        'type': link.link_type,
+                    })
 
-    type_check(
-        entities,
-        (list, tuple, set),
-        msg='`entities` must be specified and given as a list of AiiDA entities',
-    )
-    entities = list(entities)
+        # generate a list of queries to encapsulate all required entities
+        entity_queries = _collect_entity_queries(
+            node_ids_to_be_exported,
+            entities_starting_set,
+            node_pk_2_uuid_mapping,
+            include_comments,
+            include_logs,
+        )
 
-    all_fields_info, unique_identifiers = get_all_fields_info()
+        total_entities = sum(query.count() for query in entity_queries.values())
 
-    entities_starting_set, given_node_entry_ids = _get_starting_node_ids(entities)
+        # write all entity data fields
+        if total_entities:
+            exported_entity_pks = _write_entity_data(
+                total_entities=total_entities,
+                entity_queries=entity_queries,
+                writer=writer_context,
+                batch_size=batch_size
+            )
+        else:
+            exported_entity_pks = defaultdict(set)
+            EXPORT_LOGGER.info('No entities were found to export')
 
-    (
-        node_ids_to_be_exported,
-        node_pk_2_uuid_mapping,
-        links_uuid,
-        traversal_rules,
-    ) = _collect_node_ids(given_node_entry_ids, **traversal_rules)
-
-    _check_node_licenses(node_ids_to_be_exported, allowed_licenses, forbidden_licenses)
-
-    entries_queries = _collect_entity_queries(
-        node_ids_to_be_exported,
-        entities_starting_set,
-        node_pk_2_uuid_mapping,
-        include_comments,
-        include_logs,
-    )
+        # write mappings of groups to the nodes they contain
+        if exported_entity_pks[GROUP_ENTITY_NAME]:
 
-    entity_data = _perform_export_queries(entries_queries)
-
-    # note this was originally below the attributes and group_uuid gather
-    check_process_nodes_sealed({
-        node_pk for node_pk, content in entity_data.get(NODE_ENTITY_NAME, {}).items()
-        if content['node_type'].startswith('process.')
-    })
-
-    number_of_entities = sum(len(entities) for entities in entity_data.values())
-    if not number_of_entities:
-        EXPORT_LOGGER.log(msg='Nothing to store, exiting...', level=LOG_LEVEL_REPORT)
-        return None
-    EXPORT_LOGGER.log(
-        msg=(
-            f'Exporting a total of {number_of_entities} database entries, '
-            f'of which {len(node_ids_to_be_exported)} are Nodes.'
-        ),
-        level=LOG_LEVEL_REPORT,
-    )
+            EXPORT_LOGGER.debug('Writing group UUID -> [nodes UUIDs]')
 
-    groups_uuid = _get_groups_uuid(entity_data)
+            _write_group_mappings(
+                group_pks=exported_entity_pks[GROUP_ENTITY_NAME], batch_size=batch_size, writer=writer_context
+            )
 
-    # Turn sets into lists to be able to export them as JSON metadata.
-    for entity, entity_set in entities_starting_set.items():
-        entities_starting_set[entity] = list(entity_set)  # type: ignore
+        # copy all required node repositories
+        if exported_entity_pks[NODE_ENTITY_NAME]:
 
-    metadata = ArchiveMetadata(
-        export_version=EXPORT_VERSION,
-        aiida_version=get_version(),
-        unique_identifiers=unique_identifiers,
-        all_fields_info=all_fields_info,
-        graph_traversal_rules=traversal_rules,
-        entities_starting_set=entities_starting_set,
-        include_comments=include_comments,
-        include_logs=include_logs,
-    )
+            _write_node_repositories(
+                node_pks=exported_entity_pks[NODE_ENTITY_NAME],
+                node_pk_2_uuid_mapping=node_pk_2_uuid_mapping,
+                writer=writer_context
+            )
 
-    all_node_uuids = {node_pk_2_uuid_mapping[_] for _ in node_ids_to_be_exported}
+        EXPORT_LOGGER.info('Finalizing Export...')
 
-    # we get the node data last, because it is generally the largest data source
-    # and later we may look to stream this data in chunks
-    node_data = _collect_node_data(node_ids_to_be_exported)
+    # summarize export
+    export_summary = '\n  - '.join(f'{name:<6}: {len(pks)}' for name, pks in exported_entity_pks.items())
+    if exported_entity_pks:
+        EXPORT_LOGGER.info('Exported Entities:\n  - ' + export_summary + '\n')
+    # TODO
+    # EXPORT_LOGGER.info('Writer Information:\n %s', writer.export_info)
 
-    return ArchiveData(
-        metadata=metadata,
-        node_uuids=all_node_uuids,
-        group_uuids=groups_uuid,
-        link_uuids=links_uuid,
-        entity_data=entity_data,
-        node_data=node_data,
-    )
+    return writer
 
 
 def _get_starting_node_ids(entities: List[Any]) -> Tuple[DefaultDict[str, Set[str]], Set[int]]:
@@ -438,95 +360,52 @@ def _get_starting_node_ids(entities: List[Any]) -> Tuple[DefaultDict[str, Set[st
     if not total:
         return entities_starting_set, given_node_entry_ids
 
-    with get_progress_reporter()(desc='Collecting chosen entities', total=total) as progress:
-        for entry in entities:
-
-            if issubclass(entry.__class__, orm.Group):
-                entities_starting_set[GROUP_ENTITY_NAME].add(entry.uuid)
-            elif issubclass(entry.__class__, orm.Node):
-                entities_starting_set[NODE_ENTITY_NAME].add(entry.uuid)
-                given_node_entry_ids.add(entry.pk)
-            elif issubclass(entry.__class__, orm.Computer):
-                entities_starting_set[COMPUTER_ENTITY_NAME].add(entry.uuid)
-            else:
-                raise exceptions.ArchiveExportError(
-                    f'I was given {entry} ({type(entry)}),'
-                    ' which is not a Node, Computer, or Group instance'
-                )
-
-            progress.update()
-
-        # Add all the nodes contained within the specified groups
-        if GROUP_ENTITY_NAME in entities_starting_set:
-
-            progress.set_description_str('Retrieving Nodes from Groups ...', refresh=False)
-
-            # Use single query instead of given_group.nodes iterator for performance.
-            qh_groups = (
-                orm.QueryBuilder().append(
-                    orm.Group,
-                    filters={
-                        'uuid': {
-                            'in': entities_starting_set[GROUP_ENTITY_NAME]
-                        }
-                    },
-                    tag='groups',
-                ).queryhelp
-            )
-
-            node_results = (
-                orm.QueryBuilder(**qh_groups).append(orm.Node, project=['id', 'uuid'], with_group='groups').all()
+    for entry in entities:
+
+        if issubclass(entry.__class__, orm.Group):
+            entities_starting_set[GROUP_ENTITY_NAME].add(entry.uuid)
+        elif issubclass(entry.__class__, orm.Node):
+            entities_starting_set[NODE_ENTITY_NAME].add(entry.uuid)
+            given_node_entry_ids.add(entry.pk)
+        elif issubclass(entry.__class__, orm.Computer):
+            entities_starting_set[COMPUTER_ENTITY_NAME].add(entry.uuid)
+        else:
+            raise exceptions.ArchiveExportError(
+                f'I was given {entry} ({type(entry)}),'
+                ' which is not a Node, Computer, or Group instance'
             )
-            if node_results:
-                pks, uuids = map(list, zip(*node_results))
-                entities_starting_set[NODE_ENTITY_NAME].update(uuids)  # type: ignore
-                given_node_entry_ids.update(pks)  # type: ignore
-                del node_results, pks, uuids
-
-    return entities_starting_set, given_node_entry_ids
 
+    # Add all the nodes contained within the specified groups
+    if GROUP_ENTITY_NAME in entities_starting_set:
+
+        # Use single query instead of given_group.nodes iterator for performance.
+        qh_groups = (
+            orm.QueryBuilder().append(
+                orm.Group,
+                filters={
+                    'uuid': {
+                        'in': entities_starting_set[GROUP_ENTITY_NAME]
+                    }
+                },
+                tag='groups',
+            ).queryhelp
+        )
+        node_query = orm.QueryBuilder(**qh_groups).append(orm.Node, project=['id', 'uuid'], with_group='groups')
+        node_count = node_query.count()
 
-def _collect_node_ids(given_node_entry_ids: Set[int],
-                      **traversal_rules: bool) -> Tuple[Set[int], Dict[int, str], List[dict], Dict[str, bool]]:
-    """Iteratively explore the AiiDA graph to find further nodes that should also be exported
-
-    At the same time, we will create the links_uuid list of dicts to be exported
-
-    """
-    with get_progress_reporter()(desc='Traversing provenance via links ...', total=1) as progress:
-        traverse_output = get_nodes_export(starting_pks=given_node_entry_ids, get_links=True, **traversal_rules)
-        progress.update()
-
-    node_ids_to_be_exported = traverse_output['nodes']
-    graph_traversal_rules = traverse_output['rules']
+        if node_count:
+            with get_progress_reporter()(desc='Collecting nodes in groups', total=node_count) as progress:
 
-    # A utility dictionary for mapping PK to UUID.
-    if node_ids_to_be_exported:
-        qbuilder = orm.QueryBuilder().append(
-            orm.Node,
-            project=('id', 'uuid'),
-            filters={'id': {
-                'in': node_ids_to_be_exported
-            }},
-        )
-        node_pk_2_uuid_mapping = dict(qbuilder.all())
-    else:
-        node_pk_2_uuid_mapping = {}
+                pks, uuids = [], []
+                for pk, uuid in node_query.all():
+                    progress.update()
+                    pks.append(pk)
+                    uuids.append(uuid)
 
-    # The set of tuples now has to be transformed to a list of dicts
-    links_uuid = [{
-        'input': node_pk_2_uuid_mapping[link.source_id],
-        'output': node_pk_2_uuid_mapping[link.target_id],
-        'label': link.link_label,
-        'type': link.link_type,
-    } for link in traverse_output['links']]
+            entities_starting_set[NODE_ENTITY_NAME].update(uuids)  # type: ignore
+            given_node_entry_ids.update(pks)  # type: ignore
 
-    return (
-        node_ids_to_be_exported,
-        node_pk_2_uuid_mapping,
-        links_uuid,
-        graph_traversal_rules,
-    )
+    return entities_starting_set, given_node_entry_ids
 
 
 def _check_node_licenses(
@@ -551,6 +430,25 @@ def _check_node_licenses(
         check_licenses(node_licenses, allowed_licenses, forbidden_licenses)
 
 
+def _get_model_fields(entity_name: str) -> List[str]:
+    """Return a list of fields to retrieve for a particular entity
+
+    :param entity_name: name of database entity, such as Node
+
+    """
+    all_fields_info, _ = get_all_fields_info()
+    project_cols = ['id']
+    entity_prop = all_fields_info[entity_name].keys()
+    # Here we do the necessary renaming of properties
+    for prop in entity_prop:
+        # nprop contains the list of projections
+        nprop = (
+            file_fields_to_model_fields[entity_name][prop] if prop in file_fields_to_model_fields[entity_name] else prop
+        )
+        project_cols.append(nprop)
+    return project_cols
+
+
 def _collect_entity_queries(
     node_ids_to_be_exported: Set[int],
     entities_starting_set: DefaultDict[str, Set[str]],
@@ -562,12 +460,10 @@ def _collect_entity_queries(
     # pylint: disable=too-many-locals
     given_log_entry_ids = set()
     given_comment_entry_ids = set()
-    all_fields_info, _ = get_all_fields_info()
 
-    total = 1 + (((1 if include_logs else 0) + (1 if include_comments else 0)) if node_ids_to_be_exported else 0)
-    with get_progress_reporter()(desc='Initializing export of all entities', total=total) as progress:
+    total = 2 + (((1 if include_logs else 0) + (1 if include_comments else 0)) if node_ids_to_be_exported else 0)
+    with get_progress_reporter()(desc='Building entity database queries', total=total) as progress:
 
-        # Universal "entities" attributed to all types of nodes
         # Logs
         if include_logs and node_ids_to_be_exported:
             # Get related log(s) - universal for all nodes
@@ -611,31 +507,14 @@ def _collect_entity_queries(
 
         progress.update()
 
-    entities_to_add: Dict[str, orm.QueryBuilder] = {}
-    if not given_entities:
-        return entities_to_add
-
-    with get_progress_reporter()(total=len(given_entities)) as progress:
-
-        pbar_base_str = 'Preparing entities'
+        entities_to_add: Dict[str, orm.QueryBuilder] = {}
+        if not given_entities:
+            progress.update()
+            return entities_to_add
 
         for given_entity in given_entities:
-            progress.set_description_str(f'{pbar_base_str} - {given_entity}s', refresh=False)
-            progress.update()
 
-            project_cols = ['id']
-            # The following gets a list of fields that we need,
-            # e.g. user, mtime, uuid, computer
-            entity_prop = all_fields_info[given_entity].keys()
-
-            # Here we do the necessary renaming of properties
-            for prop in entity_prop:
-                # nprop contains the list of projections
-                nprop = (
-                    file_fields_to_model_fields[given_entity][prop]
-                    if prop in file_fields_to_model_fields[given_entity] else prop
-                )
-                project_cols.append(nprop)
+            project_cols = _get_model_fields(given_entity)
 
             # Getting the ids that correspond to the right entity
             entry_uuids_to_add = entities_starting_set.get(given_entity, set())
@@ -659,140 +538,120 @@ def _collect_entity_queries(
             )
             entities_to_add[given_entity] = builder
 
-    return entities_to_add
-
-
-def _perform_export_queries(entries_queries: Dict[str, orm.QueryBuilder]) -> Dict[str, Dict[int, dict]]:
-    """Start automatic recursive export data generation
-
-    :param entries_queries: partial queries for all entities to export
-
-    :return: export data mappings by entity type -> pk -> db_columns, e.g.
-        {'ENTITY_NAME': {<pk>: {'uuid': 'abc', ...}, ...}, ...}
-        Note: this data does not yet contain attributes and extras
-
-    """
-    EXPORT_LOGGER.debug('GATHERING DATABASE ENTRIES...')
+        progress.update()
 
-    all_fields_info, _ = get_all_fields_info()
+    return entities_to_add
 
-    export_data = defaultdict(dict)  # type: dict
-    if not entries_queries:
-        return export_data
 
+def _write_entity_data(
+    total_entities: int, entity_queries: Dict[str, orm.QueryBuilder], writer: ArchiveWriterAbstract, batch_size: int
+) -> Dict[str, Set[int]]:
+    """Iterate through data returned from entity queries, serialize the DB fields, then write to the export."""
+    all_fields_info, unique_identifiers = get_all_fields_info()
     entity_separator = '_'
-    counts = [p_query.count() for p_query in entries_queries.values()]
-    with get_progress_reporter()(total=sum(counts)) as progress:
 
-        for entity_name, partial_query in entries_queries.items():
+    exported_entity_pks: Dict[str, Set[int]] = defaultdict(set)
+    unsealed_node_pks: Set[int] = set()
 
-            progress.set_description_str(f'Exporting {entity_name} fields', refresh=False)
+    with get_progress_reporter()(total=total_entities, desc='Writing entity data') as progress:
 
-            foreign_fields = {k: v for k, v in all_fields_info[entity_name].items() if 'requires' in v}
+        for entity_name, entity_query in entity_queries.items():
 
+            foreign_fields = {k: v for k, v in all_fields_info[entity_name].items() if 'requires' in v}
             for value in foreign_fields.values():
                 ref_model_name = value['requires']
                 fill_in_query(
-                    partial_query,
+                    entity_query,
                     entity_name,
                     ref_model_name,
                     [entity_name],
                     entity_separator,
                 )
 
-            for temp_d in partial_query.iterdict():
+            for query_results in entity_query.iterdict(batch_size=batch_size):
 
                 progress.update()
 
-                for key in temp_d:
-                    # Get current entity
-                    current_entity = key.split(entity_separator)[-1]
+                for key, value in query_results.items():
+
+                    pk = value['id']
 
-                    # This is a empty result of an outer join.
+                    # This is an empty result of an outer join.
                     # It should not be taken into account.
-                    if temp_d[key]['id'] is None:
+                    if pk is None:
                         continue
 
-                    export_data[current_entity].update({
-                        temp_d[key]['id']:
-                        serialize_dict(
-                            temp_d[key],
-                            remove_fields=['id'],
-                            rename_fields=model_fields_to_file_fields[current_entity],
-                        )
-                    })
-
-    return export_data
+                    # Get current entity
+                    current_entity = key.split(entity_separator)[-1]
 
+                    # don't allow duplication
+                    if pk in exported_entity_pks[current_entity]:
+                        continue
 
-def _collect_node_data(all_node_pks: Set[int]) -> Iterable[Tuple[str, dict, dict]]:
-    """Gather attributes and extras for nodes
+                    exported_entity_pks[current_entity].add(pk)
 
-    :param export_data:  mappings by entity type -> pk -> db_columns
-    :param all_node_pks: set of pks
+                    fields = serialize_dict(
+                        value,
+                        remove_fields=['id'],
+                        rename_fields=model_fields_to_file_fields[current_entity],
+                    )
 
-    :return: iterable of (uuid, attributes, extras)
+                    if current_entity == NODE_ENTITY_NAME and fields['node_type'].startswith('process.'):
+                        if fields['attributes'].get('sealed', False) is not True:
+                            unsealed_node_pks.add(pk)
 
-    """
-    # ATTRIBUTES and EXTRAS
-    EXPORT_LOGGER.debug('GATHERING NODE ATTRIBUTES AND EXTRAS...')
-    node_data = []
+                    writer.write_entity_data(current_entity, pk, unique_identifiers[current_entity], fields)
 
-    # Another QueryBuilder query to get the attributes and extras.
-    if all_node_pks:
-        all_nodes_query = orm.QueryBuilder().append(
-            orm.Node,
-            filters={'id': {
-                'in': all_node_pks
-            }},
-            project=['id', 'attributes', 'extras'],
+    if unsealed_node_pks:
+        raise exceptions.ExportValidationError(
+            'All ProcessNodes must be sealed before they can be exported. '
+            f"Node(s) with PK(s): {', '.join(str(pk) for pk in unsealed_node_pks)} is/are not sealed."
         )
 
-        with get_progress_reporter()(total=all_nodes_query.count()) as progress:
-            progress.set_description_str('Exporting Attributes and Extras', refresh=False)
-
-            for node_pk, attributes, extras in all_nodes_query.iterall():
-                progress.update()
+    return exported_entity_pks
 
-                node_data.append((str(node_pk), attributes, extras))
 
-    return node_data
+def _write_group_mappings(*, group_pks: Set[int], batch_size: int, writer: ArchiveWriterAbstract):
+    """Query for node UUIDs in exported groups, and write these these mappings to the archive file."""
+    group_uuid_query = orm.QueryBuilder().append(
+        orm.Group,
+        filters={
+            'id': {
+                'in': list(group_pks)
+            }
+        },
+        project='uuid',
+        tag='groups',
+    ).append(orm.Node, project='uuid', with_group='groups')
 
+    groups_uuid_to_node_uuids = defaultdict(set)
+    for group_uuid, node_uuid in group_uuid_query.iterall(batch_size=batch_size):
+        groups_uuid_to_node_uuids[group_uuid].add(node_uuid)
 
-def _get_groups_uuid(entity_data: Dict[str, Dict[int, dict]]) -> Dict[str, Set[str]]:
-    """Get node UUIDs per group."""
-    EXPORT_LOGGER.debug('GATHERING GROUP ELEMENTS...')
-    groups_uuid: Dict[str, Set[str]] = defaultdict(set)
-    # If a group is in the exported data, we export the group/node correlation
-    if GROUP_ENTITY_NAME not in entity_data:
-        return groups_uuid
+    for group_uuid, node_uuids in groups_uuid_to_node_uuids.items():
+        writer.write_group_nodes(group_uuid, list(node_uuids))
 
-    group_uuids_with_node_uuids = (
-        orm.QueryBuilder().append(
-            orm.Group,
-            filters={
-                'id': {
-                    'in': entity_data[GROUP_ENTITY_NAME]
-                }
-            },
-            project='uuid',
-            tag='groups',
-        ).append(orm.Node, project='uuid', with_group='groups')
-    )
 
-    total_node_uuids_for_groups = group_uuids_with_node_uuids.count()
+def _write_node_repositories(
+    *, node_pks: Set[int], node_pk_2_uuid_mapping: Dict[int, str], writer: ArchiveWriterAbstract
+):
+    """Write all exported node repositories to the archive file."""
+    with get_progress_reporter()(total=len(node_pks), desc='Exporting node repositories: ') as progress:
 
-    if not total_node_uuids_for_groups:
-        return groups_uuid
+        for pk in node_pks:
 
-    with get_progress_reporter()(desc='Exporting Groups ...', total=total_node_uuids_for_groups) as progress:
+            uuid = node_pk_2_uuid_mapping[pk]
 
-        for group_uuid, node_uuid in group_uuids_with_node_uuids.iterall():
+            progress.set_description_str(f'Exporting node repositories: {pk}', refresh=False)
             progress.update()
 
-            groups_uuid[group_uuid].add(node_uuid)
-
-    return groups_uuid
+            src = RepositoryFolder(section=Repository._section_name, uuid=uuid)  # pylint: disable=protected-access
+            if not src.exists():
+                raise exceptions.ArchiveExportError(
+                    f'Unable to find the repository folder for Node with UUID={uuid} '
+                    'in the local repository'
+                )
+            writer.write_node_repo_folder(uuid, src._abspath)  # pylint: disable=protected-access
 
 
 # THESE FUNCTIONS ARE ONLY ADDED FOR BACK-COMPATIBILITY
@@ -800,10 +659,10 @@ def _get_groups_uuid(entity_data: Dict[str, Dict[int, dict]]) -> Dict[str, Set[s
 
 def export_tree(
     entities: Optional[Iterable[Any]] = None,
-    folder: Optional[Union[Folder, ZipFolder]] = None,
+    folder: Optional[Folder] = None,
     allowed_licenses: Optional[Union[list, Callable]] = None,
     forbidden_licenses: Optional[Union[list, Callable]] = None,
-    silent: bool = False,
+    silent: Optional[bool] = None,
     include_comments: bool = True,
     include_logs: bool = True,
     **traversal_rules: bool,
@@ -813,7 +672,7 @@ def export_tree(
         Support for the parameter `what` will be removed in `v2.0.0`. Please use `entities` instead.
     :param entities: a list of entity instances; they can belong to different models/entities.
 
-    :param folder: a temporary folder to build the archive before compression.
+    :param folder: a temporary folder to build the archive in.
 
     :param allowed_licenses: List or function. If a list, then checks whether all licenses of Data nodes are in the
         list. If a function, then calls function for licenses of Data nodes expecting True if license is allowed, False
@@ -881,18 +740,14 @@ def export_zip(
         'export_zip function is deprecated and will be removed in AiiDA v2.0.0, use `export` instead',
         AiidaDeprecationWarning
     )  # pylint: disable=no-member
-    report = export(
+    writer = export(
         entities=entities,
         filename=filename,
         file_format=ExportFileFormat.ZIP,
         use_compression=use_compression,
         **kwargs,
     )
-    if report.time_write_stop is not None:
-        return (report.time_collect_start, report.time_write_stop)
-
-    # there was no data to write
-    return (report.time_collect_start, report.time_collect_stop)
+    return writer.export_info['writer_entered'], writer.export_info['writer_exited']
 
 
 def export_tar(
@@ -915,18 +770,15 @@ def export_tar(
         'export_tar function is deprecated and will be removed in AiiDA v2.0.0, use `export` instead',
         AiidaDeprecationWarning
     )  # pylint: disable=no-member
-    report = export(
+    writer = export(
         entities=entities,
         filename=filename,
         file_format=ExportFileFormat.TAR_GZIPPED,
         **kwargs,
     )
 
-    if report.writer_data is not None:
-        return (
-            report.time_collect_start, report.time_write_stop, report.writer_data['compression_time_start'],
-            report.writer_data['compression_time_stop']
-        )  # type: ignore
-
-    # there was no data to write
-    return (report.time_collect_start, report.time_collect_stop, report.time_collect_stop, report.time_collect_stop)
+    # the tar is now directly written to, so no compression start/stop time!
+    return (
+        writer.export_info['writer_entered'], writer.export_info['writer_exited'], writer.export_info['writer_exited'],
+        writer.export_info['writer_exited']
+    )
diff --git a/aiida/tools/importexport/dbexport/utils.py b/aiida/tools/importexport/dbexport/utils.py
index c4445406fa..770c9354dd 100644
--- a/aiida/tools/importexport/dbexport/utils.py
+++ b/aiida/tools/importexport/dbexport/utils.py
@@ -12,7 +12,7 @@
 import warnings
 
 from aiida.orm import QueryBuilder, ProcessNode
-from aiida.common.log import AIIDA_LOGGER, LOG_LEVEL_REPORT, override_log_formatter
+from aiida.common.log import AIIDA_LOGGER, LOG_LEVEL_REPORT
 from aiida.common.warnings import AiidaDeprecationWarning
 
 from aiida.tools.importexport.common import exceptions
@@ -272,7 +272,6 @@ def check_process_nodes_sealed(nodes):
         )
 
 
-@override_log_formatter('%(message)s')
 def summary(*, file_format, export_version, outfile, include_comments, include_logs, traversal_rules):
     """Print summary for export"""
     from tabulate import tabulate
diff --git a/aiida/tools/importexport/dbimport/backends/common.py b/aiida/tools/importexport/dbimport/backends/common.py
index d12df73f43..d5e9044949 100644
--- a/aiida/tools/importexport/dbimport/backends/common.py
+++ b/aiida/tools/importexport/dbimport/backends/common.py
@@ -35,7 +35,7 @@ def _copy_node_repositories(*, uuids_to_create: List[str], reader: ArchiveReader
     if not uuids_to_create:
         return
     IMPORT_LOGGER.debug('CREATING NEW NODE REPOSITORIES...')
-    with get_progress_reporter()(total=1) as progress:
+    with get_progress_reporter()(total=1, desc='Creating new node repos') as progress:
 
         _callback = create_callback(progress)
 
diff --git a/environment.yml b/environment.yml
index 10f8ffc49f..04411ca482 100644
--- a/environment.yml
+++ b/environment.yml
@@ -8,6 +8,7 @@ dependencies:
 - python~=3.7
 - aldjemy~=0.9.1
 - alembic~=1.2
+- archive-path~=0.2.1
 - circus~=0.16.1
 - click-completion~=0.5.1
 - click-config-file~=0.6.0
diff --git a/requirements/requirements-py-3.6.txt b/requirements/requirements-py-3.6.txt
index 30d70faf69..f3fb98a8cc 100644
--- a/requirements/requirements-py-3.6.txt
+++ b/requirements/requirements-py-3.6.txt
@@ -5,6 +5,7 @@ alembic==1.4.1
 aniso8601==8.0.0
 appdirs==1.4.4
 appnope==0.1.0
+archive-path==0.2.1
 ase==3.19.0
 attrs==19.3.0
 Babel==2.8.0
diff --git a/requirements/requirements-py-3.7.txt b/requirements/requirements-py-3.7.txt
index b758303284..2cb9e68289 100644
--- a/requirements/requirements-py-3.7.txt
+++ b/requirements/requirements-py-3.7.txt
@@ -5,6 +5,7 @@ alembic==1.4.1
 aniso8601==8.0.0
 appdirs==1.4.4
 appnope==0.1.0
+archive-path==0.2.1
 ase==3.19.0
 attrs==19.3.0
 Babel==2.8.0
diff --git a/requirements/requirements-py-3.8.txt b/requirements/requirements-py-3.8.txt
index 3cc3a951a3..b83043f2d6 100644
--- a/requirements/requirements-py-3.8.txt
+++ b/requirements/requirements-py-3.8.txt
@@ -4,6 +4,7 @@ aldjemy==0.9.1
 alembic==1.4.1
 aniso8601==8.0.0
 appnope==0.1.0
+archive-path==0.2.1
 ase==3.19.0
 attrs==19.3.0
 Babel==2.8.0
diff --git a/requirements/requirements-py-3.9.txt b/requirements/requirements-py-3.9.txt
index 583bc2247e..5404a60c2f 100644
--- a/requirements/requirements-py-3.9.txt
+++ b/requirements/requirements-py-3.9.txt
@@ -3,6 +3,7 @@ alabaster==0.7.12
 aldjemy==0.9.1
 alembic==1.4.3
 aniso8601==8.0.0
+archive-path==0.2.1
 ase==3.20.1
 attrs==20.2.0
 Babel==2.8.0
diff --git a/setup.json b/setup.json
index ba298ce6ec..13c4e02ae8 100644
--- a/setup.json
+++ b/setup.json
@@ -23,6 +23,7 @@
     "install_requires": [
         "aldjemy~=0.9.1",
         "alembic~=1.2",
+        "archive-path~=0.2.1",
         "circus~=0.16.1",
         "click-completion~=0.5.1",
         "click-config-file~=0.6.0",
diff --git a/tests/cmdline/commands/test_export.py b/tests/cmdline/commands/test_export.py
index dc490b921d..4b2cfa16bf 100644
--- a/tests/cmdline/commands/test_export.py
+++ b/tests/cmdline/commands/test_export.py
@@ -153,7 +153,7 @@ def test_migrate_versions_old(self):
             filename_output = next(tempfile._get_candidate_names())  # pylint: disable=protected-access
 
             try:
-                options = [filename_input, filename_output]
+                options = ['--verbosity', 'DEBUG', filename_input, filename_output]
                 result = self.cli_runner.invoke(cmd_export.migrate, options)
                 self.assertIsNone(result.exception, result.output)
                 self.assertTrue(os.path.isfile(filename_output))
diff --git a/tests/tools/importexport/migration/conftest.py b/tests/tools/importexport/migration/conftest.py
index 73d9857546..fcf7d2807b 100644
--- a/tests/tools/importexport/migration/conftest.py
+++ b/tests/tools/importexport/migration/conftest.py
@@ -11,9 +11,10 @@
 import tarfile
 import zipfile
 
+from archive_path import TarPath, ZipPath
 import pytest
 
-from aiida.tools.importexport.archive import safe_extract_tar, safe_extract_zip, CacheFolder
+from aiida.tools.importexport.archive import CacheFolder
 from aiida.tools.importexport.archive.migrations.utils import verify_metadata_version
 from tests.utils.archives import get_archive_file
 
@@ -54,9 +55,9 @@ def _migrate(filename_archive, version_old, version_new, migration_method, archi
         out_path = tmp_path / 'out.aiida'
 
         if zipfile.is_zipfile(archive_path):
-            safe_extract_zip(archive_path, out_path)
+            ZipPath(archive_path).extract_tree(out_path)
         elif tarfile.is_tarfile(archive_path):
-            safe_extract_tar(archive_path, out_path)
+            TarPath(archive_path).extract_tree(out_path)
         else:
             raise ValueError('invalid file format, expected either a zip archive or gzipped tarball')
 
diff --git a/tests/tools/importexport/migration/test_v03_to_v04.py b/tests/tools/importexport/migration/test_v03_to_v04.py
index 4eaebcb7cd..a6406f26d0 100644
--- a/tests/tools/importexport/migration/test_v03_to_v04.py
+++ b/tests/tools/importexport/migration/test_v03_to_v04.py
@@ -12,9 +12,11 @@
 import tarfile
 import zipfile
 
+from archive_path import TarPath, ZipPath
+
 from aiida.common.exceptions import NotExistent
 from aiida.common import json
-from aiida.tools.importexport.archive import safe_extract_tar, safe_extract_zip, CacheFolder
+from aiida.tools.importexport.archive import CacheFolder
 from aiida.tools.importexport.archive.migrations.utils import verify_metadata_version
 from aiida.tools.importexport.archive.migrations.v03_to_v04 import migrate_v3_to_v4
 
@@ -31,9 +33,9 @@ def test_migrate_external(external_archive, tmp_path):
 
     # Migrate
     if zipfile.is_zipfile(dirpath_archive):
-        safe_extract_zip(dirpath_archive, out_path)
+        ZipPath(dirpath_archive).extract_tree(out_path)
     elif tarfile.is_tarfile(dirpath_archive):
-        safe_extract_tar(dirpath_archive, out_path)
+        TarPath(dirpath_archive).extract_tree(out_path)
     else:
         raise ValueError('invalid file format, expected either a zip archive or gzipped tarball')
 
@@ -320,9 +322,9 @@ def test_illegal_create_links(external_archive, tmp_path):
 
     # Migrate
     if zipfile.is_zipfile(dirpath_archive):
-        safe_extract_zip(dirpath_archive, out_path)
+        ZipPath(dirpath_archive).extract_tree(out_path)
     elif tarfile.is_tarfile(dirpath_archive):
-        safe_extract_tar(dirpath_archive, out_path)
+        TarPath(dirpath_archive).extract_tree(out_path)
     else:
         raise ValueError('invalid file format, expected either a zip archive or gzipped tarball')
 
diff --git a/tests/tools/importexport/test_simple.py b/tests/tools/importexport/test_simple.py
index 409c11411b..2255f714ea 100644
--- a/tests/tools/importexport/test_simple.py
+++ b/tests/tools/importexport/test_simple.py
@@ -8,176 +8,168 @@
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
 """Simple tests for the export and import routines"""
-import os
-import shutil
 import tarfile
-import tempfile
+
+import pytest
 
 from aiida import orm
-from aiida.backends.testbase import AiidaTestCase
 from aiida.common import json
 from aiida.common.exceptions import LicensingException
+from aiida.common.folders import SandboxFolder
+from aiida.common.links import LinkType
 from aiida.tools.importexport import import_data, export
 from aiida.tools.importexport.common import exceptions
 
-from tests.utils.configuration import with_temp_dir
-
-
-class TestSimple(AiidaTestCase):
-    """Test simple ex-/import cases"""
-
-    def setUp(self):
-        self.reset_database()
-
-    def tearDown(self):
-        self.reset_database()
-
-    @with_temp_dir
-    def test_base_data_nodes(self, temp_dir):
-        """Test ex-/import of Base Data nodes"""
-        # producing values for each base type
-        values = ('Hello', 6, -1.2399834e12, False)  # , ["Bla", 1, 1e-10])
-        filename = os.path.join(temp_dir, 'export.aiida')
-
-        # producing nodes:
-        nodes = [cls(val).store() for val, cls in zip(values, (orm.Str, orm.Int, orm.Float, orm.Bool))]
-        # my uuid - list to reload the node:
-        uuids = [n.uuid for n in nodes]
-        # exporting the nodes:
-        export(nodes, filename=filename, silent=True)
-        # cleaning:
-        self.clean_db()
-        self.create_user()
-        # Importing back the data:
-        import_data(filename, silent=True)
-        # Checking whether values are preserved:
-        for uuid, refval in zip(uuids, values):
-            self.assertEqual(orm.load_node(uuid).value, refval)
-
-    @with_temp_dir
-    def test_calc_of_structuredata(self, temp_dir):
-        """Simple ex-/import of CalcJobNode with input StructureData"""
-        from aiida.common.links import LinkType
-
-        struct = orm.StructureData()
-        struct.store()
-
-        calc = orm.CalcJobNode()
-        calc.computer = self.computer
-        calc.set_option('resources', {'num_machines': 1, 'num_mpiprocs_per_machine': 1})
-
-        calc.add_incoming(struct, link_type=LinkType.INPUT_CALC, link_label='link')
-        calc.store()
-        calc.seal()
-
-        pks = [struct.pk, calc.pk]
-
-        attrs = {}
-        for pk in pks:
-            node = orm.load_node(pk)
-            attrs[node.uuid] = dict()
-            for k in node.attributes.keys():
-                attrs[node.uuid][k] = node.get_attribute(k)
-
-        filename = os.path.join(temp_dir, 'export.aiida')
-
-        export([calc], filename=filename, silent=True)
-
-        self.clean_db()
-        self.create_user()
-
-        # NOTE: it is better to load new nodes by uuid, rather than assuming
-        # that they will have the first 3 pks. In fact, a recommended policy in
-        # databases is that pk always increment, even if you've deleted elements
-        import_data(filename, silent=True)
-        for uuid in attrs:
-            node = orm.load_node(uuid)
-            for k in attrs[uuid].keys():
-                self.assertEqual(attrs[uuid][k], node.get_attribute(k))
-
-    def test_check_for_export_format_version(self):
-        """Test the check for the export format version."""
-        # Creating a folder for the archive files
-        export_file_tmp_folder = tempfile.mkdtemp()
-        unpack_tmp_folder = tempfile.mkdtemp()
-        try:
-            struct = orm.StructureData()
-            struct.store()
-
-            filename = os.path.join(export_file_tmp_folder, 'export.aiida')
-            export([struct], filename=filename, file_format='tar.gz', silent=True)
-
-            with tarfile.open(filename, 'r:gz', format=tarfile.PAX_FORMAT) as tar:
-                tar.extractall(unpack_tmp_folder)
-
-            with open(os.path.join(unpack_tmp_folder, 'metadata.json'), 'r', encoding='utf8') as fhandle:
-                metadata = json.load(fhandle)
-            metadata['export_version'] = 0.0
-
-            with open(os.path.join(unpack_tmp_folder, 'metadata.json'), 'wb') as fhandle:
-                json.dump(metadata, fhandle)
-
-            with tarfile.open(filename, 'w:gz', format=tarfile.PAX_FORMAT) as tar:
-                tar.add(unpack_tmp_folder, arcname='')
-
-            self.tearDownClass()
-            self.setUpClass()
-
-            with self.assertRaises(exceptions.IncompatibleArchiveVersionError):
-                import_data(filename, silent=True)
-        finally:
-            # Deleting the created temporary folders
-            shutil.rmtree(export_file_tmp_folder, ignore_errors=True)
-            shutil.rmtree(unpack_tmp_folder, ignore_errors=True)
-
-    def test_control_of_licenses(self):
-        """Test control of licenses."""
-        from aiida.common.folders import SandboxFolder
-        from aiida.tools.importexport.dbexport import export_tree
-
-        struct = orm.StructureData()
-        struct.source = {'license': 'GPL'}
-        struct.store()
-
-        folder = SandboxFolder()
-        export_tree([struct], folder=folder, silent=True, allowed_licenses=['GPL'])
-        # Folder should contain two files of metadata + nodes/
-        self.assertEqual(len(folder.get_content_list()), 3)
-
-        folder = SandboxFolder()
-        export_tree([struct], folder=folder, silent=True, forbidden_licenses=['Academic'])
-        # Folder should contain two files of metadata + nodes/
-        self.assertEqual(len(folder.get_content_list()), 3)
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, allowed_licenses=['CC0'])
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, forbidden_licenses=['GPL'])
-
-        def cc_filter(license_):
-            return license_.startswith('CC')
-
-        def gpl_filter(license_):
-            return license_ == 'GPL'
-
-        def crashing_filter():
-            raise NotImplementedError('not implemented yet')
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, allowed_licenses=cc_filter)
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, forbidden_licenses=gpl_filter)
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, allowed_licenses=crashing_filter)
-
-        folder = SandboxFolder()
-        with self.assertRaises(LicensingException):
-            export_tree([struct], folder=folder, silent=True, forbidden_licenses=crashing_filter)
+
+@pytest.mark.parametrize('file_format', ('zip', 'tar.gz'))
+def test_base_data_nodes(aiida_profile, tmp_path, file_format):
+    """Test ex-/import of Base Data nodes"""
+    aiida_profile.reset_db()
+
+    # producing values for each base type
+    values = ('Hello', 6, -1.2399834e12, False)  # , ["Bla", 1, 1e-10])
+    filename = str(tmp_path / 'export.aiida')
+
+    # producing nodes:
+    nodes = [cls(val).store() for val, cls in zip(values, (orm.Str, orm.Int, orm.Float, orm.Bool))]
+    # my uuid - list to reload the node:
+    uuids = [n.uuid for n in nodes]
+    # exporting the nodes:
+    export(nodes, filename=filename, file_format=file_format)
+    # cleaning:
+    aiida_profile.reset_db()
+    # Importing back the data:
+    import_data(filename)
+    # Checking whether values are preserved:
+    for uuid, refval in zip(uuids, values):
+        assert orm.load_node(uuid).value == refval
+
+
+@pytest.mark.parametrize('file_format', ('zip', 'tar.gz'))
+def test_calc_of_structuredata(aiida_profile, tmp_path, file_format):
+    """Simple ex-/import of CalcJobNode with input StructureData"""
+    aiida_profile.reset_db()
+
+    struct = orm.StructureData()
+    struct.store()
+
+    computer = orm.Computer(
+        label='localhost-test',
+        description='localhost computer set up by test manager',
+        hostname='localhost-test',
+        workdir=str(tmp_path / 'workdir'),
+        transport_type='local',
+        scheduler_type='direct'
+    )
+    computer.store()
+    computer.configure()
+
+    calc = orm.CalcJobNode()
+    calc.computer = computer
+    calc.set_option('resources', {'num_machines': 1, 'num_mpiprocs_per_machine': 1})
+
+    calc.add_incoming(struct, link_type=LinkType.INPUT_CALC, link_label='link')
+    calc.store()
+    calc.seal()
+
+    pks = [struct.pk, calc.pk]
+
+    attrs = {}
+    for pk in pks:
+        node = orm.load_node(pk)
+        attrs[node.uuid] = dict()
+        for k in node.attributes.keys():
+            attrs[node.uuid][k] = node.get_attribute(k)
+
+    filename = str(tmp_path / 'export.aiida')
+
+    export([calc], filename=filename, file_format=file_format)
+
+    aiida_profile.reset_db()
+
+    import_data(filename)
+    for uuid in attrs:
+        node = orm.load_node(uuid)
+        for k in attrs[uuid].keys():
+            assert attrs[uuid][k] == node.get_attribute(k)
+
+
+def test_check_for_export_format_version(aiida_profile, tmp_path):
+    """Test the check for the export format version."""
+    # Creating a folder for the archive files
+    export_file_tmp_folder = tmp_path / 'export_tmp'
+    export_file_tmp_folder.mkdir()
+    unpack_tmp_folder = tmp_path / 'unpack_tmp'
+    unpack_tmp_folder.mkdir()
+
+    aiida_profile.reset_db()
+
+    struct = orm.StructureData()
+    struct.store()
+
+    filename = str(export_file_tmp_folder / 'export.aiida')
+    export([struct], filename=filename, file_format='tar.gz')
+
+    with tarfile.open(filename, 'r:gz', format=tarfile.PAX_FORMAT) as tar:
+        tar.extractall(unpack_tmp_folder)
+
+    with (unpack_tmp_folder / 'metadata.json').open('r', encoding='utf8') as fhandle:
+        metadata = json.load(fhandle)
+    metadata['export_version'] = 0.0
+
+    with (unpack_tmp_folder / 'metadata.json').open('wb') as fhandle:
+        json.dump(metadata, fhandle)
+
+    with tarfile.open(filename, 'w:gz', format=tarfile.PAX_FORMAT) as tar:
+        tar.add(unpack_tmp_folder, arcname='')
+
+    aiida_profile.reset_db()
+
+    with pytest.raises(exceptions.IncompatibleArchiveVersionError):
+        import_data(filename)
+
+
+@pytest.mark.usefixtures('clear_database_before_test')
+def test_control_of_licenses():
+    """Test control of licenses."""
+    struct = orm.StructureData()
+    struct.source = {'license': 'GPL'}
+    struct.store()
+
+    folder = SandboxFolder()
+    export([struct], file_format='folder', writer_init={'folder': folder}, allowed_licenses=['GPL'])
+    # Folder should contain two files of metadata + nodes/
+    assert len(folder.get_content_list()) == 3
+
+    folder.erase(create_empty_folder=True)
+    assert len(folder.get_content_list()) == 0
+    export([struct], file_format='folder', writer_init={'folder': folder}, forbidden_licenses=['Academic'])
+    # Folder should contain two files of metadata + nodes/
+    assert len(folder.get_content_list()) == 3
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', allowed_licenses=['CC0'])
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', forbidden_licenses=['GPL'])
+
+    def cc_filter(license_):
+        return license_.startswith('CC')
+
+    def gpl_filter(license_):
+        return license_ == 'GPL'
+
+    def crashing_filter():
+        raise NotImplementedError('not implemented yet')
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', allowed_licenses=cc_filter)
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', forbidden_licenses=gpl_filter)
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', allowed_licenses=crashing_filter)
+
+    with pytest.raises(LicensingException):
+        export([struct], file_format='null', forbidden_licenses=crashing_filter)
diff --git a/tests/tools/importexport/test_specific_import.py b/tests/tools/importexport/test_specific_import.py
index cd195f0275..874a296e57 100644
--- a/tests/tools/importexport/test_specific_import.py
+++ b/tests/tools/importexport/test_specific_import.py
@@ -198,7 +198,7 @@ def test_missing_node_repo_folder_export(self, temp_dir):
         # Try to export, check it raises and check the raise message
         filename = os.path.join(temp_dir, 'export.aiida')
         with self.assertRaises(exceptions.ArchiveExportError) as exc:
-            export([node], filename=filename, silent=True)
+            export([node], filename=filename)
 
         self.assertIn(f'Unable to find the repository folder for Node with UUID={node_uuid}', str(exc.exception))
         self.assertFalse(os.path.exists(filename), msg='The archive file should not exist')
@@ -258,9 +258,9 @@ def test_missing_node_repo_folder_import(self, temp_dir):
 
         # Try to import, check it raises and check the raise message
         with self.assertRaises(exceptions.CorruptArchive) as exc:
-            import_data(filename_corrupt, silent=True)
+            import_data(filename_corrupt)
 
-        self.assertIn(f'Unable to find the repository folder for Node with UUID={node_uuid}', str(exc.exception))
+        self.assertIn('Unable to find required folder in archive', str(exc.exception))
 
     @with_temp_dir
     def test_empty_repo_folder_export(self, temp_dir):
@@ -292,16 +292,16 @@ def test_empty_repo_folder_export(self, temp_dir):
             'zip archive': os.path.join(temp_dir, 'export.zip')
         }
 
-        export_tree([node], folder=Folder(archive_variants['archive folder']), silent=True)
-        export([node], filename=archive_variants['tar archive'], file_format='tar.gz', silent=True)
-        export([node], filename=archive_variants['zip archive'], file_format='zip', silent=True)
+        export_tree([node], folder=Folder(archive_variants['archive folder']))
+        export([node], filename=archive_variants['tar archive'], file_format='tar.gz')
+        export([node], filename=archive_variants['zip archive'], file_format='zip')
 
         for variant, filename in archive_variants.items():
             self.reset_database()
             node_count = orm.QueryBuilder().append(orm.Dict, project='uuid').count()
             self.assertEqual(node_count, 0, msg=f'After DB reset {node_count} Dict Nodes was (wrongly) found')
 
-            import_data(filename, silent=True)
+            import_data(filename)
             builder = orm.QueryBuilder().append(orm.Dict, project='uuid')
             imported_node_count = builder.count()
             self.assertEqual(
diff --git a/tests/utils/archives.py b/tests/utils/archives.py
index 2777be3796..f683154fa1 100644
--- a/tests/utils/archives.py
+++ b/tests/utils/archives.py
@@ -14,8 +14,9 @@
 import zipfile
 from typing import List
 
+from archive_path import read_file_in_tar, read_file_in_zip
+
 from aiida.common import json
-from aiida.tools.importexport.archive.common import read_file_in_tar, read_file_in_zip
 from tests.static import STATIC_DIR