Repository: sentinelsat/sentinelsat (language: python)

Issue:
Downloading all products failed for unknown reason
While downloading a good bounce of S1 images, some downloads fail with the error in the subject, with no more information about  the issue (it seems yet another not managed LTA error). For instance:
```
['d78ede38-b22b-43af-9db0-8472403195d5']
Fetching archival status: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  2.02 products/s]
Downloading all products failed for unknown reason
['4d079c46-712d-462d-be18-be0e92828f35']
Fetching archival status: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:01<00:00,  1.19s/ products]
Downloading all products failed for unknown reason

```
Downloads with download_all() cannot be interrupted
When using sentinelsat interactively in a notebook or REPL the downloads started with `download_all()` run in a background thread and are not interrupted by a `KeyboardInterrupt`. This is less than ideal when working with large 5 GB products, for example.

`concurrent.futures` does not offer a straightforward solution for killing the background threads, unfortunately - https://stackoverflow.com/questions/42782953/python-concurrent-futures-how-to-make-it-cancelable


Discussion / hints:

Fix patch:
diff --git a/.codecov.yml b/.codecov.yml
index 4a298e92..da690c1b 100644
--- a/.codecov.yml
+++ b/.codecov.yml
@@ -2,7 +2,7 @@ coverage:
   status:
     project:
       default:
-        target: 85%
+        target: 82%
     patch:
       default:
         target: 70%
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index f988d207..e7fb4643 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -5,26 +5,49 @@ All notable changes to ``sentinelsat`` will be listed here.
 
 [Unreleased] – YYYY-MM-DD
 -------------------------
+``download_all()`` has been completely rewritten to correctly handle server-side concurrency limits and bugs. 
 
 CLI changes
 ~~~~~~~~~~~
-* New CLI option to specify the format of the output report (`#526 <https://github.com/sentinelsat/sentinelsat/pull/526>`_ `@avalentino <https://github.com/avalentino>`_
+* New CLI option ``--fmt`` to specify the format of the listed products. (`#526 <https://github.com/sentinelsat/sentinelsat/pull/526>`_ `@avalentino <https://github.com/avalentino>`_
+* More detailed progress info during downloads. Added progressbars for overall download and LTA retrieval progress.
+  Product names are now included in individual download progress bars. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
 * Progressbars no longer conflict with logging output. (`#531 <https://github.com/sentinelsat/sentinelsat/issues/531>`_ `@valgur <https://github.com/valgur>`_)
+* Added ``--fail-fast`` option for downloading. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
 
 Added
 ~~~~~
+* Added ``concurrent_dl_limit`` and ``concurrent_lta_trigger_limit`` parameters to ``SentinelAPI`` which are used to ensure that
+  the server-side limit for concurrent downloads (4 for scihub.copernicus.eu) and the maximum number of concurrent LTA retrievals
+  (set to 10 by default) are respected everywhere. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* Added an optional ``lta_timeout`` parameter to ``SentinelAPI``, which sets the maximum time to wait for products to be retrieved from the LTA.
+  Defaults to unlimited. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* Added ``dl_retry_delay`` parameter to ``SentinelAPI``, which defaults to 10 seconds and limits the rate of download retries to give the server some time to recover. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* Added ``fail_fast`` option to ``download_all()``. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* Added support for SHA3-256 checksums used by some newer Sentinel-2 products. (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
+* For more advanced use-cases, ``Downloader.download_all()``, which returns a detailed ``DownloadStatus`` enum for each product, can now be used. (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
 * GeoJSON geometries consisting of multiple features are now supported. (`#530 <https://github.com/sentinelsat/sentinelsat/issues/530>`_ `@mackland <https://github.com/mackland>`_)
-* Better handling 403 error in ``trigger_offline_retrieval()`` (`#491 <https://github.com/sentinelsat/sentinelsat/issues/491>`_ `@z4zz <https://github.com/z4zz>`_)
-* Added support for SHA3-256 checksums. (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
 
 Changed
 ~~~~~~~
+* ``download_all()`` now waits for all LTA retrievals to finish before exiting. (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
+* Default ``n_concurrent_dl`` in ``download_all()`` increased from 2 to 4. (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
+* Default LTA retry period was reduced from 600 to 60 seconds. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* ``SentinelProductsAPI`` has been merged into ``SentinelAPI`` and will be removed in a future release.  (`#523 <https://github.com/sentinelsat/sentinelsat/issues/523>`_ `@valgur <https://github.com/valgur>`_)
 * Suppressed the unnecessary server stacktrace printing for ``QueryLengthError``. (`#524 <https://github.com/sentinelsat/sentinelsat/issues/524>`_ `@valgur <https://github.com/valgur>`_)
 
 Fixed
 ~~~~~
 * Changed logic for ``_get_filename`` when Content-Disposition header not present (`#494 <https://github.com/sentinelsat/sentinelsat/issues/494>`_ `@rbrishabh <https://github.com/rbrishabh>`_)
-* Fixed online status checks accidentally counting towards the concurrent downloads limit on the server, causing downloads to fail. (`#508 <https://github.com/sentinelsat/sentinelsat/issues/508>`_ `@avalentino <https://github.com/avalentino>`_).
+* ``download_all()`` strictly tracks and limits the number of concurrent GET requests made (which includes LTA triggering) to respect server-side limits. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_, `#508 <https://github.com/sentinelsat/sentinelsat/issues/508>`_ `@avalentino <https://github.com/avalentino>`_).
+* Triggering is now much more robust against server-side bugs (such as the frequent HTTP 500: NullPonterException, Postgres bugs, etc.). (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
+* Better handling of HTTP 403 error in ``trigger_offline_retrieval()`` (`#491 <https://github.com/sentinelsat/sentinelsat/issues/491>`_ `@z4zz <https://github.com/z4zz>`_)
+* Downloading with ``download_all()`` can now be cleanly interrupted without the downloads remaining active in the background. (`#481, #532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_) 
+
+Development Changes
+~~~~~~~~~~~~~~~~~~~
+* All downloading-related functionality has been moved into a separate ``Downloader`` class to keep the code more organized.
+  The existing API has been kept as-is, however. (`#532 <https://github.com/sentinelsat/sentinelsat/issues/532>`_ `@valgur <https://github.com/valgur>`_)
 
 [1.0.1] – 2021-05-25
 -------------------------
diff --git a/README.rst b/README.rst
index de3e402d..dad616f4 100644
--- a/README.rst
+++ b/README.rst
@@ -213,6 +213,10 @@ Options
      - --download
      -  
      - Download all results of the query.
+   * -
+     - --fail-fast
+     -
+     - Skip all other other downloads if one fails.
    * -  
      - --path
      - PATH
diff --git a/docs/api_overview.rst b/docs/api_overview.rst
index 5fcc9635..a9ac2c6f 100644
--- a/docs/api_overview.rst
+++ b/docs/api_overview.rst
@@ -264,6 +264,8 @@ Copernicus Open Access Hub's quota currently permits users to request an offline
 A product's availability can be checked with a regular OData query by evaluating the ``Online`` property value
 or by using the :meth:`~sentinel.SentinelAPI.is_online()` convenience method.
 
+The retrieval of offline products from the LTA can be triggered using :meth:`~sentinel.SentinelAPI.trigger_offline_retrieval()`.
+
 .. code-block:: python
 
     product_info = api.get_product_odata(<product_id>)
@@ -272,33 +274,17 @@ or by using the :meth:`~sentinel.SentinelAPI.is_online()` convenience method.
     is_online = api.is_online(<product_id>)
 
     if is_online:
-        print('Product {} is online. Starting download.'.format(<product_id>))
+        print(f'Product {<product_id>} is online. Starting download.')
         api.download(<product_id>)
     else:
-        print('Product {} is not online.'.format(<product_id>))
-
-When trying to download an offline product with :meth:`~sentinel.SentinelAPI.download` it will trigger its retrieval from the LTA.
-
-Given a list of offline and online products, :meth:`~sentinel.SentinelAPI.download_all` will download online products, while concurrently triggering the retrieval of offline products from the LTA.
-Offline products that become online while downloading will be added to the download queue.
-:meth:`~sentinel.SentinelAPI.download_all` terminates when the download queue is empty, even if not all products were retrieved from the LTA.
-We suggest repeatedly calling :meth:`~sentinel.SentinelAPI.download_all` to download all products, either manually or using a third-party library, e.g. `tenacity <https://github.com/jd/tenacity>`_.
-
-
-.. code-block:: python
-
-    from sentinelsat import SentinelAPI
-    import tenacity
-
-    api = SentinelAPI('user', 'password')
-
-    @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed(3600))
-    def download_all(*args, **kwargs):
-        return api.download_all(*args, **kwargs)
-
-    downloaded, triggered, failed = download_all(<product_ids>)
+        print(f'Product {<product_id>} is not online.')
+        api.trigger_offline_retrieval(<product_id>)
 
+When trying to download an offline product with :meth:`~sentinel.SentinelAPI.download` it will trigger its retrieval from the LTA and raise an ``LTATriggered`` exception.
 
+Given a list of offline and online products, :meth:`~sentinel.SentinelAPI.download_all` will download online products, while concurrently triggering the retrieval of offline products from the LTA in the background.
+:meth:`~sentinel.SentinelAPI.download_all` terminates when all products have been retrieved from the LTA and downloaded.
+If you wish to avoid the possibly lengthy retrieval process of offline products, you can either set the :attr:`~sentinel.SentinelAPI.lta_timeout` parameter to a low value or filter the list of products based on the :meth:`~sentinel.SentinelAPI.is_online()` status beforehand.
 
 Logging
 -------
@@ -331,33 +317,34 @@ or add a custom handler for :mod:`sentinelsat` (as implemented in `cli.py`)
   logger.addHandler(h)
 
 
-Product node API
-----------------
+Downloading parts of products
+-----------------------------
 
-The product node API is implemented in the :mod:`sentinelsat.products`
-module and, in addition to all the features provided by the standard API,
-allows to download only some of the files included in a Sentinel product
-exploiting the `node selection feature`_ provided by the OData_ Web API.
+Both :meth:`~sentinel.SentinelAPI.download` and :meth:`~sentinel.SentinelAPI.download_all`
+include a ``nodefilter`` parameter that can be used to specify a subset of files within the product
+that should be downloaded, skipping the rest. 
+The downloaded files will be written to disk as individual files instead of a single archive file.
+This functionality makes use of the `node selection feature`_ of the OData_ Web API.
 
 .. code-block:: python
 
-  from sentinelsat import SentinelProductsAPI, make_path_filter
+  from sentinelsat import SentinelAPI, make_path_filter
 
   # define the filter function to select files (to be excluded in this case)
-  nodefilter = make_path_filter("*measurement/*", exclude=True)
+  path_filter = make_path_filter("*measurement/*", exclude=True)
 
   # connect to the API
   api = SentinelProductAPI("user", "password")
 
   # download a single product excluding measurement files
-  api.download(<product_id>, nodefilter=nodefilter)
+  api.download(<product_id>, nodefilter=path_filter)
 
-Of course it also works for multiple products:
+Of course it also works for multiple products, too:
 
 .. code-block:: python
 
   # download a multiple products excluding measurement files
-  api.download_all(<products>, nodefilter=nodefilter)
+  api.download_all(<products>, nodefilter=path_filter)
 
 The example above downloads all files in each of the requested products only
 *excluding* (large) measurements files.
@@ -366,9 +353,9 @@ This can be useful for analyses exclusively based on product annotations
 included in the product.
 
 The file selection is implemented by specifying a *nodefilter* function that
-is called for each file (only excluding the manifest which is always downloaded)
-in the requested products and returns `True` if the file have to be downloaded,
-`False` otherwise.
+is called for each file (except for the manifest, which is always downloaded)
+in the requested products and returns ``True`` if the file have to be downloaded,
+``False`` otherwise.
 
 The *nodefilter* function has the following signature:
 
@@ -383,11 +370,11 @@ keys:
 :url:
     the URL to download the product file node
 :node_path:
-    the file *path* within the product (e.g. "./preview/map-overlay.kml")
+    the *path* within the product (e.g. "./preview/map-overlay.kml")
 :size:
     the file size in bytes (int)
 :md5:
-    the file md5 checksum
+    the file's MD5 checksum
 
 In the example above it has been used an helper function
 (:func:`sentinelsat.products.make_path_filter`), provided by the
@@ -398,7 +385,7 @@ The following code:
 
 .. code-block:: python
 
-  nodefilter = make_path_filter("*measurement/*", exclude=True)
+  path_filter = make_path_filter("*measurement/*", exclude=True)
 
 is more or less equivalent to:
 
@@ -406,11 +393,8 @@ is more or less equivalent to:
 
   import fnmatch
 
-  def node_filter(node_info):
-    if not fnmatch.fnmatch(node_info["node_path"].lower(), pattern):
-      return True
-    else:
-      return False
+  def path_filter(node_info):
+    return fnmatch.fnmatch(node_info["node_path"].lower(), pattern)
 
 The :mod:`sentinelsat` product node API also provides:
 
@@ -421,7 +405,7 @@ The :mod:`sentinelsat` product node API also provides:
   This function can be used to download the entire Sentinel product as a
   directory instead of downloading a single zip archive.
 
-Of course the user can write its own *nodefilter* functions if necessary.
+Of course the user can write their own *nodefilter* functions if necessary.
 
 
 .. _`node selection feature`: https://scihub.copernicus.eu/twiki/do/view/SciHubUserGuide/ODataAPI#Discover_Product_Nodes
@@ -480,17 +464,13 @@ measurement files for sub-swaths "EW1" and "EW2":
 
 .. code-block:: python
 
-  nodefilter = make_path_filter("*s1?-ew[12]-slc-hh-*.tiff")
+  path_filter = make_path_filter("*s1?-ew[12]-slc-hh-*.tiff")
 
 Considering that e.g. a Dual Pol Extended Wide Swath Sentinel-1 product
 includes 2 measurement files for each of the 5 sub-swath the above filter
 allows to reduce consistently the amount of data to be downloaded
 (form 10 to 2 TIFF files approx 700MB each).
 
-Of course the *nodefilter* function have to be used to initialize the
-product node API :class:`sentinelsat.products.SentinelProductsAPI` as
-explained above.
-
 .. code-block:: python
 
-    api.download_all(<products>, nodefilter=nodefilter)
+    api.download_all(<products>, nodefilter=path_filter)
diff --git a/docs/cli.rst b/docs/cli.rst
index a9e2ab6c..1f782a95 100644
--- a/docs/cli.rst
+++ b/docs/cli.rst
@@ -170,6 +170,10 @@ Options
 
     Download all results of the query.
 
+.. option:: --fail-fast
+
+    Skip all other other downloads if one fails.
+
 .. option:: --path <directory>
 
     Set the directory where the files will be saved.
diff --git a/docs/common_issues.rst b/docs/common_issues.rst
index 4aae3f00..701e9881 100644
--- a/docs/common_issues.rst
+++ b/docs/common_issues.rst
@@ -22,7 +22,7 @@ SciHub servers are known to have outages due to high demand, try again later.
 
 .. rubric:: Query fails with 'Longitude/Latitude is out of bounds, check your JSON format or data.'
 
-Standard GeoJSON specification contains only WGS84 format, check if your data complies with it.
+The server expects coordinates in the WGS84 format, check if your data complies with it.
 
 .. rubric:: A query to Eumetsat CODA fails with *HTTP 401 - Full authentication is required* even with the correct password.
 
@@ -40,6 +40,6 @@ the decimal point.
 
 .. rubric:: Anything else?
 
-Make sure to check the `issues on GitHub`__ too.
+Make sure to check the existing `issues on GitHub`__ too.
 
 __ https://github.com/sentinelsat/sentinelsat/issues?q=is%3Aissue
diff --git a/sentinelsat/__init__.py b/sentinelsat/__init__.py
index 3f8b7d2f..5db14d65 100644
--- a/sentinelsat/__init__.py
+++ b/sentinelsat/__init__.py
@@ -20,10 +20,8 @@
     read_geojson,
     placename_to_wkt,
 )
-
-from .products import (
-    SentinelProductsAPI,
-    make_path_filter,
-    make_size_filter,
-    all_nodes_filter,
+from .download import (
+    Downloader,
+    DownloadStatus,
 )
+from .products import all_nodes_filter, make_path_filter, make_size_filter, SentinelProductsAPI
diff --git a/sentinelsat/download.py b/sentinelsat/download.py
new file mode 100644
index 00000000..e487dd99
--- /dev/null
+++ b/sentinelsat/download.py
@@ -0,0 +1,802 @@
+import concurrent.futures
+import enum
+import itertools
+import shutil
+import threading
+import time
+import traceback
+from collections import namedtuple
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing
+from pathlib import Path
+from typing import Any, Dict
+from xml.etree import ElementTree as etree
+
+from sentinelsat.exceptions import (
+    InvalidChecksumError,
+    LTAError,
+    LTATriggered,
+    SentinelAPIError,
+    ServerError,
+    UnauthorizedError,
+)
+
+
+class DownloadStatus(enum.Enum):
+    """Status info for ``Downloader.download_all()``."""
+
+    UNAVAILABLE = enum.auto()
+    OFFLINE = enum.auto()
+    TRIGGERED = enum.auto()
+    ONLINE = enum.auto()
+    DOWNLOAD_STARTED = enum.auto()
+    DOWNLOADED = enum.auto()
+
+    def __bool__(self):
+        return self == DownloadStatus.DOWNLOADED
+
+
+class Downloader:
+    def __init__(
+        self,
+        api,
+        *,
+        node_filter=None,
+        verify_checksum=True,
+        fail_fast=False,
+        n_concurrent_dl=None,
+        max_attempts=10,
+        dl_retry_delay=10,
+        lta_retry_delay=60,
+        lta_timeout=None
+    ):
+        """
+        Manages downloading of products or parts of them.
+
+        Intended for internal use, but may also be used directly if more fine-grained
+        configuration or custom download logic is needed.
+
+        Parameters
+        ----------
+        api : SentinelAPI
+            A SentinelAPI instance.
+        node_filter : callable, optional
+            The callable is used to select which files of each product will be downloaded.
+            If None (the default), the full products will be downloaded.
+            See ``sentinelsat.products`` for sample node filters.
+        verify_checksum : bool, default True
+            If True, verify the downloaded files' integrity by checking its checksum.
+            Throws InvalidChecksumError if the checksum does not match.
+        fail_fast : bool, default False
+            if True, all other downloads are cancelled when one of the downloads fails in ``download_all()``.
+        n_concurrent_dl : integer, optional
+            Number of concurrent downloads.
+            Defaults to the maximum allowed by ``SentinelAPI.concurrent_dl_limit``.
+        max_attempts : int, default 10
+            Number of allowed retries before giving up downloading a product in ``download_all()``.
+        dl_retry_delay : float, default 10
+            Number of seconds to wait between retrying of failed downloads.
+        lta_retry_delay : float, default 60
+            Number of seconds to wait between requests to the Long Term Archive.
+        lta_timeout : float, optional
+            Maximum number of seconds to wait for triggered products to come online.
+            Defaults to no timeout.
+        """
+        from sentinelsat import SentinelAPI
+
+        self.api: SentinelAPI = api
+        self.logger = self.api.logger
+        self._tqdm = self.api._tqdm
+
+        self.node_filter = node_filter
+        self.verify_checksum = verify_checksum
+        self.fail_fast = fail_fast
+        self.max_attempts = max_attempts
+        self.n_concurrent_dl = n_concurrent_dl or self.api.concurrent_dl_limit
+        self.dl_retry_delay = dl_retry_delay
+        self.lta_retry_delay = lta_retry_delay
+        self.lta_timeout = lta_timeout
+        self.chunk_size = 2 ** 20  # download in 1 MB chunks by default
+
+    def download(self, id, directory=".", *, stop_event=None):
+        """Download a product.
+
+        Parameters
+        ----------
+        id : string
+            UUID of the product, e.g. 'a8dd0cfd-613e-45ce-868c-d79177b916ed'
+        directory : string or Path, optional
+            Where the file will be downloaded
+        stop_event : threading.Event, optional
+            An event object can be provided and set to interrupt the download.
+
+        Returns
+        -------
+        product_info : dict
+            Dictionary containing the product's info from get_product_odata() as well as
+            the path on disk.
+
+        Raises
+        ------
+        InvalidChecksumError
+            If the checksum does not match the checksum on the server.
+        LTATriggered
+            If the product has been archived and its retrieval was successfully triggered.
+        LTAError
+            If the product has been archived and its retrieval failed.
+        """
+        if self.node_filter:
+            return self._download_with_node_filter(id, directory, stop_event)
+
+        product_info = self.api.get_product_odata(id)
+        filename = self.api._get_filename(product_info)
+        path = Path(directory) / filename
+        product_info["path"] = str(path)
+        product_info["downloaded_bytes"] = 0
+
+        if path.exists():
+            # We assume that the product has been downloaded and is complete
+            return product_info
+
+        # An incomplete download triggers the retrieval from the LTA if the product is not online
+        if not self.api.is_online(id):
+            self.trigger_offline_retrieval(id)
+            raise LTATriggered(id)
+
+        self._download_common(product_info, path, stop_event)
+        return product_info
+
+    def _download_with_node_filter(self, id, directory, stop_event):
+        product_info = self.api.get_product_odata(id)
+        product_path = Path(directory) / (product_info["title"] + ".SAFE")
+        product_info["node_path"] = "./" + product_info["title"] + ".SAFE"
+        manifest_path = product_path / "manifest.safe"
+        if not manifest_path.exists() and self.trigger_offline_retrieval(id):
+            raise LTATriggered(id)
+        manifest_info, _ = self.api._get_manifest(product_info, manifest_path)
+        product_info["nodes"] = {
+            manifest_info["node_path"]: manifest_info,
+        }
+        node_infos = self._filter_nodes(manifest_path, product_info, self.node_filter)
+        product_info["nodes"].update(node_infos)
+        for node_info in node_infos.values():
+            if stop_event and stop_event.is_set():
+                raise concurrent.futures.CancelledError()
+            node_path = node_info["node_path"]
+            path = (product_path / node_path).resolve()
+            node_info["path"] = path
+            node_info["downloaded_bytes"] = 0
+
+            self.logger.debug("Downloading %s node to %s", id, path)
+            self.logger.debug("Node URL for %s: %s", id, node_info["url"])
+
+            if path.exists():
+                # We assume that the product node has been downloaded and is complete
+                continue
+            self._download_common(node_info, path, stop_event)
+        return product_info
+
+    def _download_common(self, product_info: Dict[str, Any], path: Path, stop_event):
+        # Use a temporary file for downloading
+        temp_path = path.with_name(path.name + ".incomplete")
+        skip_download = False
+        if temp_path.exists():
+            size = temp_path.stat().st_size
+            if size > product_info["size"]:
+                self.logger.warning(
+                    "Existing incomplete file %s is larger than the expected final size"
+                    " (%s vs %s bytes). Deleting it.",
+                    str(temp_path),
+                    size,
+                    product_info["size"],
+                )
+                temp_path.unlink()
+            elif size == product_info["size"]:
+                if self.verify_checksum and not self.api._checksum_compare(temp_path, product_info):
+                    # Log a warning since this should never happen
+                    self.logger.warning(
+                        "Existing incomplete file %s appears to be fully downloaded but "
+                        "its checksum is incorrect. Deleting it.",
+                        str(temp_path),
+                    )
+                    temp_path.unlink()
+                else:
+                    skip_download = True
+            else:
+                # continue downloading
+                self.logger.info(
+                    "Download will resume from existing incomplete file %s.", temp_path
+                )
+                pass
+        if not skip_download:
+            # Store the number of downloaded bytes for unit tests
+            temp_path.parent.mkdir(parents=True, exist_ok=True)
+            product_info["downloaded_bytes"] = self._download(
+                product_info["url"],
+                temp_path,
+                product_info["size"],
+                path.name,
+                stop_event,
+            )
+        # Check integrity with MD5 checksum
+        if self.verify_checksum is True:
+            if not self.api._checksum_compare(temp_path, product_info):
+                temp_path.unlink()
+                raise InvalidChecksumError("File corrupt: checksums do not match")
+        # Download successful, rename the temporary file to its proper name
+        shutil.move(temp_path, path)
+        return product_info
+
+    def download_all(self, products, directory="."):
+        """Download a list of products.
+
+        Parameters
+        ----------
+        products : list
+            List of product IDs
+        directory : string or Path, optional
+            Directory where the files will be downloaded
+
+        Raises
+        ------
+        By default, raises the most recent downloading exception if all downloads failed.
+        If ``self.fail_fast`` is set to True, raises the encountered exception on the first failed
+        download instead.
+
+        Returns
+        -------
+        dict[string, DownloadStatus]
+            The status of all products.
+        dict[string, Exception]
+            Exception info for any failed products.
+        dict[string, dict]
+            A dictionary containing the product information for each product
+            (unless the product was unavailable).
+        """
+
+        ResultTuple = namedtuple("ResultTuple", ["statuses", "exceptions", "product_infos"])
+        product_ids = list(set(products))
+        assert self.n_concurrent_dl > 0
+        if len(product_ids) == 0:
+            return ResultTuple({}, {}, {})
+        self.logger.info(
+            "Will download %d products using %d workers", len(product_ids), self.n_concurrent_dl
+        )
+
+        statuses, online_prods, offline_prods, product_infos, exceptions = self._init_statuses(
+            product_ids
+        )
+
+        # Skip already downloaded files.
+        # Although the download method also checks, we do not need to retrieve such
+        # products from the LTA and use up our quota.
+        self._skip_existing_products(directory, offline_prods, product_infos, statuses)
+
+        stop_event = threading.Event()
+        dl_tasks = {}
+        trigger_tasks = {}
+
+        # Two separate threadpools for downloading and triggering of retrieval.
+        # Otherwise triggering might take up all threads and nothing is downloaded.
+        dl_count = len(online_prods) + len(offline_prods)
+        dl_executor = ThreadPoolExecutor(
+            max_workers=max(1, min(self.n_concurrent_dl, dl_count)),
+            thread_name_prefix="dl",
+        )
+        dl_progress = self._tqdm(
+            total=dl_count,
+            desc="Downloading products",
+            unit="product",
+        )
+        if offline_prods:
+            trigger_executor = ThreadPoolExecutor(
+                max_workers=min(self.api.concurrent_lta_trigger_limit, len(offline_prods)),
+                thread_name_prefix="trigger",
+            )
+            trigger_progress = self._tqdm(
+                total=len(offline_prods),
+                desc="LTA retrieval",
+                unit="product",
+                leave=True,
+            )
+        try:
+            # First all online products are downloaded. Subsequently, offline products that might
+            # have become available in the meantime are requested.
+            for pid in itertools.chain(online_prods, offline_prods):
+                future = dl_executor.submit(
+                    self._download_online_retry,
+                    product_infos[pid],
+                    directory,
+                    statuses,
+                    exceptions,
+                    stop_event,
+                )
+                dl_tasks[future] = pid
+
+            for pid in offline_prods:
+                future = trigger_executor.submit(
+                    self._trigger_and_wait,
+                    pid,
+                    stop_event,
+                    statuses,
+                )
+                trigger_tasks[future] = pid
+
+            for task in concurrent.futures.as_completed(list(trigger_tasks) + list(dl_tasks)):
+                pid = trigger_tasks.get(task) or dl_tasks[task]
+                exception = exceptions.get(pid)
+                if task.cancelled():
+                    exception = concurrent.futures.CancelledError()
+                if task.exception():
+                    exception = task.exception()
+
+                if task in dl_tasks:
+                    if not exception:
+                        product_infos[pid] = task.result()
+                        statuses[pid] = DownloadStatus.DOWNLOADED
+                    dl_progress.update()
+                    # Keep the LTA progress fresh
+                    if offline_prods:
+                        trigger_progress.update(0)
+                else:
+                    trigger_progress.update()
+                    if all(t.done() for t in trigger_tasks):
+                        trigger_progress.close()
+
+                if exception:
+                    exceptions[pid] = exception
+                    if self.fail_fast:
+                        raise exception from None
+                    else:
+                        self.logger.error("%s failed: %s", pid, _format_exception(exception))
+        except:
+            stop_event.set()
+            for t in list(trigger_tasks) + list(dl_tasks):
+                t.cancel()
+            raise
+        finally:
+            dl_executor.shutdown()
+            dl_progress.close()
+            if offline_prods:
+                trigger_executor.shutdown()
+                trigger_progress.close()
+
+        if not any(statuses):
+            if not exceptions:
+                raise SentinelAPIError("Downloading all products failed for an unknown reason")
+            exception = list(exceptions)[0]
+            raise exception
+
+        # Update Online status in product_infos
+        for pid, status in statuses.items():
+            if status in [DownloadStatus.OFFLINE, DownloadStatus.TRIGGERED]:
+                product_infos[pid]["Online"] = False
+            elif status != DownloadStatus.UNAVAILABLE:
+                product_infos[pid]["Online"] = True
+
+        return ResultTuple(statuses, exceptions, product_infos)
+
+    def _init_statuses(self, product_ids):
+        statuses = {pid: DownloadStatus.UNAVAILABLE for pid in product_ids}
+        online_prods = set()
+        offline_prods = set()
+        product_infos = {}
+        exceptions = {}
+        # Get online status and product info.
+        for pid in self._tqdm(
+            iterable=product_ids, desc="Fetching archival status", unit="product", delay=2
+        ):
+            assert isinstance(pid, str)
+            try:
+                info = self.api.get_product_odata(pid)
+            except UnauthorizedError:
+                raise
+            except SentinelAPIError as e:
+                exceptions[pid] = e
+                if self.fail_fast:
+                    raise
+                self.logger.error(
+                    "Getting product info for %s failed, can't download: %s",
+                    pid,
+                    _format_exception(e),
+                )
+                continue
+            product_infos[pid] = info
+            if product_infos[pid]["Online"]:
+                statuses[pid] = DownloadStatus.ONLINE
+                online_prods.add(pid)
+            else:
+                statuses[pid] = DownloadStatus.OFFLINE
+                offline_prods.add(pid)
+        return statuses, online_prods, offline_prods, product_infos, exceptions
+
+    def _skip_existing_products(self, directory, products, product_infos, statuses):
+        for pid in list(products):
+            product_info = product_infos[pid]
+            filename = self.api._get_filename(product_info)
+            path = Path(directory) / filename
+            if path.exists():
+                self.logger.info("Skipping already downloaded %s.", filename)
+                product_info["path"] = str(path)
+                statuses[pid] = DownloadStatus.DOWNLOADED
+                products.remove(pid)
+            else:
+                self.logger.info(
+                    "%s (%s) is in the LTA and retrieval will be triggered.",
+                    product_info["title"],
+                    pid,
+                )
+
+    def trigger_offline_retrieval(self, uuid):
+        """Triggers retrieval of an offline product.
+
+        Parameters
+        ----------
+        uuid : string
+            UUID of the product
+
+        Returns
+        -------
+        bool
+            True, if the product retrieval was successfully triggered.
+            False, if the product is already online.
+
+        Raises
+        ------
+        LTAError
+            If the request was not accepted due to exceeded user quota or server overload.
+        ServerError
+            If an unexpected response was received from server.
+        UnauthorizedError
+            If the provided credentials were invalid.
+
+        Notes
+        -----
+        https://scihub.copernicus.eu/userguide/LongTermArchive
+        """
+        # Request just a single byte to avoid accidental downloading of the whole product.
+        # Requesting zero bytes results in NullPointerException in the server.
+        with self.api.dl_limit_semaphore:
+            r = self.api.session.get(
+                self.api._get_download_url(uuid), headers={"Range": "bytes=0-1"}
+            )
+        cause = r.headers.get("cause-message")
+        # check https://scihub.copernicus.eu/userguide/LongTermArchive#HTTP_Status_codes
+        if r.status_code in (200, 206):
+            self.logger.debug("Product is online")
+            return False
+        elif r.status_code == 202:
+            self.logger.debug("Accepted for retrieval")
+            return True
+        elif r.status_code == 403 and cause and "concurrent flows" in cause:
+            # cause: 'An exception occured while creating a stream: Maximum number of 4 concurrent flows achieved by the user "username""'
+            self.logger.debug("Product is online but concurrent downloads limit was exceeded")
+            return False
+        elif r.status_code == 403:
+            # cause: 'User 'username' offline products retrieval quota exceeded (20 fetches max) trying to fetch product PRODUCT_FILENAME (BYTES_COUNT bytes compressed)'
+            msg = f"User quota exceeded: {cause}"
+            self.logger.error(msg)
+            raise LTAError(msg, r)
+        elif r.status_code == 503:
+            msg = f"Request not accepted: {cause}"
+            self.logger.error(msg)
+            raise LTAError(msg, r)
+        elif r.status_code < 400:
+            msg = f"Unexpected response {r.status_code}: {cause}"
+            self.logger.error(msg)
+            raise ServerError(msg, r)
+        self.api._check_scihub_response(r, test_json=False)
+
+    def get_stream(self, id, **kwargs):
+        """Exposes requests response ready to stream product to e.g. S3.
+
+        Parameters
+        ----------
+        id : string
+            UUID of the product, e.g. 'a8dd0cfd-613e-45ce-868c-d79177b916ed'
+        **kwargs
+            Any additional parameters for ``requests.get()``
+
+        Raises
+        ------
+        LTATriggered
+            If the product has been archived and its retrieval was successfully triggered.
+        LTAError
+            If the product has been archived and its retrieval failed.
+
+        Returns
+        -------
+        requests.Response:
+            Opened response object
+        """
+        if not self.api.is_online(id):
+            self.trigger_offline_retrieval(id)
+            raise LTATriggered(id)
+        with self.api.dl_limit_semaphore:
+            r = self.api.session.get(self.api._get_download_url(id), stream=True, **kwargs)
+        self.api._check_scihub_response(r, test_json=False)
+        return r
+
+    def download_quicklook(self, id, directory="."):
+        """Download a quicklook for a product.
+
+        Parameters
+        ----------
+        id : string
+            UUID of the product, e.g. 'a8dd0cfd-613e-45ce-868c-d79177b916ed'
+        directory : string or Path, optional
+            Where the image will be downloaded. Defaults to ".".
+
+        Returns
+        -------
+        quicklook_info : dict
+            Dictionary containing the quicklooks's response headers as well as the path on disk.
+        """
+        product_info = self.api.get_product_odata(id)
+        url = product_info["quicklook_url"]
+
+        path = Path(directory) / "{}.jpeg".format(product_info["title"])
+        product_info["path"] = str(path)
+        product_info["downloaded_bytes"] = 0
+        product_info["error"] = ""
+
+        self.logger.info("Downloading quicklook %s to %s", product_info["title"], path)
+
+        with self.api.dl_limit_semaphore:
+            r = self.api.session.get(url)
+        self.api._check_scihub_response(r, test_json=False)
+
+        product_info["quicklook_size"] = len(r.content)
+
+        if path.exists():
+            return product_info
+
+        content_type = r.headers["content-type"]
+        if content_type != "image/jpeg":
+            product_info["error"] = "Quicklook is not jpeg but {}".format(content_type)
+
+        if product_info["error"] == "":
+            with open(path, "wb") as fp:
+                fp.write(r.content)
+                product_info["downloaded_bytes"] = len(r.content)
+
+        return product_info
+
+    def download_all_quicklooks(self, products, directory="."):
+        """Download quicklook for a list of products.
+
+        Parameters
+        ----------
+        products : dict
+            Dict of product IDs, product data
+        directory : string or Path, optional
+            Directory where the downloaded images will be downloaded
+        n_concurrent_dl : integer
+            Number of concurrent downloads
+
+        Returns
+        -------
+        dict[string, dict]
+            A dictionary containing the return value from download_quicklook() for each
+            successfully downloaded quicklook
+        dict[string, dict]
+            A dictionary containing the error of products where either
+            quicklook was not available or it had an unexpected content type
+        """
+
+        self.logger.info("Will download %d quicklooks", len(products))
+
+        downloaded_quicklooks = {}
+        failed_quicklooks = {}
+
+        with ThreadPoolExecutor(max_workers=self.n_concurrent_dl) as dl_exec:
+            dl_tasks = {}
+            for pid in products:
+                future = dl_exec.submit(self.download_quicklook, pid, directory)
+                dl_tasks[future] = pid
+
+            completed_tasks = concurrent.futures.as_completed(dl_tasks)
+
+            for future in completed_tasks:
+                product_info = future.result()
+                if product_info["error"] == "":
+                    downloaded_quicklooks[dl_tasks[future]] = product_info
+                else:
+                    failed_quicklooks[dl_tasks[future]] = product_info["error"]
+
+        ResultTuple = namedtuple("ResultTuple", ["downloaded", "failed"])
+        return ResultTuple(downloaded_quicklooks, failed_quicklooks)
+
+    def _trigger_and_wait(self, uuid, stop_event, statuses):
+        """Continuously triggers retrieval of offline products
+
+        This function is supposed to be called in a separate thread. By setting stop_event it can be stopped.
+        """
+        with self.api.lta_limit_semaphore:
+            t0 = time.time()
+            while True:
+                if stop_event.is_set():
+                    raise concurrent.futures.CancelledError()
+                if self.lta_timeout and time.time() - t0 >= self.lta_timeout:
+                    raise LTAError(
+                        f"LTA retrieval for {uuid} timed out (lta_timeout={self.lta_timeout} seconds)"
+                    )
+                try:
+                    if self.api.is_online(uuid):
+                        break
+                    if statuses[uuid] == DownloadStatus.OFFLINE:
+                        # Trigger
+                        triggered = self.trigger_offline_retrieval(uuid)
+                        if triggered:
+                            statuses[uuid] = DownloadStatus.TRIGGERED
+                            self.logger.info(
+                                "%s accepted for retrieval, waiting for it to come online...", uuid
+                            )
+                        else:
+                            # Product is online
+                            break
+                except (LTAError, ServerError) as e:
+                    self.logger.info(
+                        "%s retrieval was not accepted: %s. Retrying in %d seconds",
+                        uuid,
+                        e.msg,
+                        self.lta_retry_delay,
+                    )
+                _wait(stop_event, self.lta_retry_delay)
+            self.logger.info("%s retrieval from LTA completed", uuid)
+            statuses[uuid] = DownloadStatus.ONLINE
+
+    def _download_online_retry(self, product_info, directory, statuses, exceptions, stop_event):
+        """Thin wrapper around download with retrying and checking whether a product is online
+
+        Parameters
+        ----------
+        product_info : dict
+        directory : string, optional
+        statuses : dict of DownloadStatus
+        exceptions : dict of Exception
+        stop_event : threading.Event
+        """
+        if self.max_attempts <= 0:
+            return
+
+        uuid = product_info["id"]
+        title = product_info["title"]
+
+        # Wait for the triggering and retrieval to complete first
+        while (
+            statuses[uuid] != DownloadStatus.ONLINE
+            and uuid not in exceptions
+            and not stop_event.is_set()
+        ):
+            _wait(stop_event, 1)
+        if uuid in exceptions:
+            return
+
+        last_exception = None
+        for cnt in range(self.max_attempts):
+            if stop_event.is_set():
+                raise concurrent.futures.CancelledError()
+            try:
+                if cnt > 0:
+                    _wait(stop_event, self.dl_retry_delay)
+                statuses[uuid] = DownloadStatus.DOWNLOAD_STARTED
+                return self.download(uuid, directory, stop_event=stop_event)
+            except (concurrent.futures.CancelledError, KeyboardInterrupt, SystemExit):
+                raise
+            except Exception as e:
+                if isinstance(e, InvalidChecksumError):
+                    self.logger.warning(
+                        "Invalid checksum. The downloaded file for '%s' is corrupted.",
+                        title,
+                    )
+                else:
+                    self.logger.exception("There was an error downloading %s", title)
+                retries_remaining = self.max_attempts - cnt - 1
+                if retries_remaining > 0:
+                    self.logger.info(
+                        "%d retries left, retrying in %s seconds...",
+                        retries_remaining,
+                        self.dl_retry_delay,
+                    )
+                else:
+                    self.logger.info("Downloading %s failed. No retries left.", title)
+                last_exception = e
+        raise last_exception
+
+    def _download(self, url, path, file_size, title, stop_event):
+        headers = {}
+        continuing = path.exists()
+        if continuing:
+            already_downloaded_bytes = path.stat().st_size
+            headers = {"Range": "bytes={}-".format(already_downloaded_bytes)}
+        else:
+            already_downloaded_bytes = 0
+        downloaded_bytes = 0
+        with self.api.dl_limit_semaphore:
+            r = self.api.session.get(url, stream=True, headers=headers)
+        with self._tqdm(
+            desc=f"Downloading {title}",
+            total=file_size,
+            unit="B",
+            unit_scale=True,
+            initial=already_downloaded_bytes,
+        ) as progress, closing(r):
+            self.api._check_scihub_response(r, test_json=False)
+            mode = "ab" if continuing else "wb"
+            with open(path, mode) as f:
+                iterator = r.iter_content(chunk_size=self.chunk_size)
+                while True:
+                    if stop_event and stop_event.is_set():
+                        raise concurrent.futures.CancelledError()
+                    try:
+                        with self.api.dl_limit_semaphore:
+                            chunk = next(iterator)
+                    except StopIteration:
+                        break
+                    if chunk:  # filter out keep-alive new chunks
+                        f.write(chunk)
+                        progress.update(len(chunk))
+                        downloaded_bytes += len(chunk)
+            # Return the number of bytes downloaded
+            return downloaded_bytes
+
+    def _dataobj_to_node_info(self, dataobj_info, product_info):
+        path = dataobj_info["href"]
+        if path.startswith("./"):
+            path = path[2:]
+
+        node_info = product_info.copy()
+        node_info["url"] = self.api._path_to_url(product_info, path, "value")
+        node_info["size"] = dataobj_info["size"]
+        if "md5" in dataobj_info:
+            node_info["md5"] = dataobj_info["md5"]
+        if "sha3-256" in dataobj_info:
+            node_info["sha3-256"] = dataobj_info["sha3-256"]
+        node_info["node_path"] = dataobj_info["href"]
+        # node_info["parent"] = product_info
+
+        return node_info
+
+    def _filter_nodes(self, manifest, product_info, nodefilter=None):
+        nodes = {}
+        xmldoc = etree.parse(manifest)
+        data_obj_section_elem = xmldoc.find("dataObjectSection")
+        for elem in data_obj_section_elem.iterfind("dataObject"):
+            dataobj_info = _xml_to_dataobj_info(elem)
+            node_info = self._dataobj_to_node_info(dataobj_info, product_info)
+            if nodefilter is not None and not nodefilter(node_info):
+                continue
+            node_path = node_info["node_path"]
+            nodes[node_path] = node_info
+        return nodes
+
+
+def _xml_to_dataobj_info(element):
+    assert etree.iselement(element)
+    assert element.tag == "dataObject"
+    data = dict(
+        id=element.attrib["ID"],
+    )
+    elem = element.find("byteStream")
+    # data["mime_type"] = elem.attrib['mimeType']
+    data["size"] = int(elem.attrib["size"])
+    elem = element.find("byteStream/fileLocation")
+    data["href"] = elem.attrib["href"]
+    # data['locator_type'] = elem.attrib["locatorType"]
+    # assert data['locator_type'] == "URL"
+
+    elem = element.find("byteStream/checksum")
+    assert elem.attrib["checksumName"].upper() in ["MD5", "SHA3-256"]
+    data[elem.attrib["checksumName"].lower()] = elem.text
+
+    return data
+
+
+def _format_exception(ex):
+    return "".join(traceback.TracebackException.from_exception(ex).format())
+
+
+def _wait(event, timeout):
+    """Wraps event.wait so it can be disabled for testing."""
+    return event.wait(timeout)
diff --git a/sentinelsat/products.py b/sentinelsat/products.py
index a827e336..00e48512 100644
--- a/sentinelsat/products.py
+++ b/sentinelsat/products.py
@@ -1,237 +1,20 @@
 import fnmatch
-from pathlib import Path
-from xml.etree import ElementTree as etree
+import warnings
 
-import sentinelsat
-from sentinelsat.exceptions import LTATriggered, SentinelAPIError
-from sentinelsat.sentinel import _check_scihub_response
+from sentinelsat.sentinel import SentinelAPI
 
 
-class SentinelProductsAPI(sentinelsat.SentinelAPI):
-    """Class to connect to Copernicus Open Access Hub, search and download imagery.
+class SentinelProductsAPI(SentinelAPI):
+    """SentinelProductsAPI functionality has been merged into SentinelAPI and Downloader
+    and will be removed in a future release."""
 
-    The products node interface allows to filter and download individual product
-    files by means of a (optional) *nodefilter* callable function.
-    For each file in the product (only excluding the manifest) the *nodefilter* function
-    is called to decide if the corresponding file must be downloaded or not.
-
-    The *nodefilter* function has the following signature::
-
-      def nodefilter(node_info: dict) -> bool:
-          ...
-
-    The *node_info* parameter is a dictionary containing information like
-
-    * the file *path* within the product (e.g. "./preview/map-overlay.kml")
-    * the file size in bytes (int)
-    * the file md5
-
-    It the *nodefilter* function returns True the corresponding file is downloaded,
-    otherwise the file is not downloaded.
-
-
-    Parameters
-    ----------
-    user : string
-        username for DataHub
-        set to None to use ~/.netrc
-    password : string
-        password for DataHub
-        set to None to use ~/.netrc
-    api_url : string, optional
-        URL of the DataHub
-        defaults to 'https://apihub.copernicus.eu/apihub'
-    show_progressbars : bool
-        Whether progressbars should be shown or not, e.g. during download. Defaults to True.
-    timeout : float or tuple, optional
-        How long to wait for DataHub response (in seconds).
-        Tuple (connect, read) allowed.
-
-    Attributes
-    ----------
-    session : requests.Session
-        Session to connect to DataHub
-    api_url : str
-        URL to the DataHub
-    page_size : int
-        Number of results per query page.
-        Current value: 100 (maximum allowed on ApiHub)
-    timeout : float or tuple
-        How long to wait for DataHub response (in seconds).
-
-
-    .. versionadded:: 0.15
-    """
-
-    def _path_to_url(self, product_info, path, urltype=None):
-        id = product_info["id"]
-        title = product_info["title"]
-        path = "/".join(["Nodes('{}')".format(item) for item in path.split("/")])
-        if urltype == "value":
-            urltype = "/$value"
-        elif urltype == "json":
-            urltype = "?$format=json"
-        elif urltype == "full":
-            urltype = "?$format=json&$expand=Attributes"
-        elif urltype is None:
-            urltype = ""
-        # else: pass urltype as is
-        return self._get_odata_url(id, f"/Nodes('{title}.SAFE')/{path}{urltype}")
-
-    def _get_manifest(self, product_info, path=None):
-        path = Path(path) if path else None
-        url = self._path_to_url(product_info, "manifest.safe", "value")
-        node_info = product_info.copy()
-        node_info["url"] = url
-        node_info["node_path"] = "./manifest.safe"
-        del node_info["md5"]
-
-        if path and path.exists():
-            self.logger.info("manifest file already available (%r), skip download", path)
-            data = path.read_bytes()
-            node_info["size"] = len(data)
-            return node_info, data
-
-        url = self._path_to_url(product_info, "manifest.safe", "json")
-        response = self.session.get(url, auth=self.session.auth)
-        _check_scihub_response(response)
-        info = response.json()["d"]
-
-        node_info["size"] = int(info["ContentLength"])
-
-        response = self.session.get(node_info["url"], auth=self.session.auth)
-        _check_scihub_response(response, test_json=False)
-        data = response.content
-        if len(data) != node_info["size"]:
-            raise SentinelAPIError("File corrupt: data length do not match")
-
-        if path:
-            path.parent.mkdir(parents=True, exist_ok=True)
-            path.write_bytes(data)
-
-        return node_info, data
-
-    def _dataobj_to_node_info(self, dataobj_info, product_info):
-        path = dataobj_info["href"]
-        if path.startswith("./"):
-            path = path[2:]
-
-        node_info = product_info.copy()
-        node_info["url"] = self._path_to_url(product_info, path, "value")
-        node_info["size"] = dataobj_info["size"]
-        if "md5" in dataobj_info:
-            node_info["md5"] = dataobj_info["md5"]
-        if "sha3-256" in dataobj_info:
-            node_info["sha3-256"] = dataobj_info["sha3-256"]
-        node_info["node_path"] = dataobj_info["href"]
-        # node_info["parent"] = product_info
-
-        return node_info
-
-    def _filter_nodes(self, manifest, product_info, nodefilter=None):
-        nodes = {}
-        xmldoc = etree.parse(manifest)
-        data_obj_section_elem = xmldoc.find("dataObjectSection")
-        for elem in data_obj_section_elem.iterfind("dataObject"):
-            dataobj_info = _xml_to_dataobj_info(elem)
-            node_info = self._dataobj_to_node_info(dataobj_info, product_info)
-            if nodefilter is not None and not nodefilter(node_info):
-                continue
-            node_path = node_info["node_path"]
-            nodes[node_path] = node_info
-        return nodes
-
-    def download(self, id, directory_path=".", checksum=True, nodefilter=None, **kwargs):
-        """Download a product.
-
-        Uses the filename on the server for the downloaded files, e.g.
-        "S1A_EW_GRDH_1SDH_20141003T003840_20141003T003920_002658_002F54_4DD1.SAFE/manifest.safe".
-
-        Incomplete downloads are continued and complete files are skipped.
-
-        Parameters
-        ----------
-        id : string
-            UUID of the product, e.g. 'a8dd0cfd-613e-45ce-868c-d79177b916ed'
-        directory_path : string, optional
-            Where the file will be downloaded
-        checksum : bool, optional
-            If True, verify the downloaded file's integrity by checking its MD5 checksum.
-            Throws InvalidChecksumError if the checksum does not match.
-            Defaults to True.
-        nodefilter : callable, optional
-            The *nodefilter* callable used to select which file of each product have to
-            be downloaded.
-            If *nodefilter* is None then no file filtering is performed and the class
-            behaves exactly as :class:`sentinelsat.sentinel.SentinelAPI`.
-
-
-        Returns
-        -------
-        product_info : dict
-            Dictionary containing the product's info from get_product_odata() as well as
-            the path on disk.
-
-        Raises
-        ------
-        InvalidChecksumError
-            If the MD5 checksum does not match the checksum on the server.
-        """
-        if nodefilter is None:
-            return sentinelsat.SentinelAPI.download(self, id, directory_path, checksum, **kwargs)
-
-        product_info = self.get_product_odata(id)
-        product_path = Path(directory_path) / (product_info["title"] + ".SAFE")
-        product_info["node_path"] = "./" + product_info["title"] + ".SAFE"
-        manifest_path = product_path / "manifest.safe"
-        if not manifest_path.exists() and self.trigger_offline_retrieval(id):
-            raise LTATriggered(id)
-
-        manifest_info, _ = self._get_manifest(product_info, manifest_path)
-        product_info["nodes"] = {
-            manifest_info["node_path"]: manifest_info,
-        }
-
-        node_infos = self._filter_nodes(manifest_path, product_info, nodefilter)
-        product_info["nodes"].update(node_infos)
-
-        for node_info in node_infos.values():
-            node_path = node_info["node_path"]
-            path = (product_path / node_path).resolve()
-            node_info["path"] = path
-            node_info["downloaded_bytes"] = 0
-
-            self.logger.info("Downloading %s node to %s", id, path)
-            self.logger.debug("Node URL for %s: %s", id, node_info["url"])
-
-            if path.exists():
-                # We assume that the product node has been downloaded and is complete
-                continue
-
-            self._download_outer(node_info, path, checksum)
-
-        return product_info
-
-
-def _xml_to_dataobj_info(element):
-    assert etree.iselement(element)
-    assert element.tag == "dataObject"
-    data = dict(
-        id=element.attrib["ID"],
-    )
-    elem = element.find("byteStream")
-    # data["mime_type"] = elem.attrib['mimeType']
-    data["size"] = int(elem.attrib["size"])
-    elem = element.find("byteStream/fileLocation")
-    data["href"] = elem.attrib["href"]
-    # data['locator_type'] = elem.attrib["locatorType"]
-    # assert data['locator_type'] == "URL"
-
-    elem = element.find("byteStream/checksum")
-    assert elem.attrib["checksumName"].upper() in ["MD5", "SHA3-256"]
-    data[elem.attrib["checksumName"].lower()] = elem.text
-
-    return data
+    def __init__(self, *args, **kwargs):
+        warnings.warn(
+            "SentinelProductsAPI has been deprecated and will be removed in a future release",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(*args, **kwargs)
 
 
 def make_size_filter(max_size):
@@ -269,7 +52,7 @@ def node_filter(node_info):
 def all_nodes_filter(node_info):
     """Node filter function to download all files.
 
-    This function can be used to download Sentinel product as a directory
+    This function can be used to download a Sentinel product as a directory
     instead of downloading a single zip archive.
 
     .. versionadded:: 0.15
diff --git a/sentinelsat/scripts/cli.py b/sentinelsat/scripts/cli.py
index 9ed2c635..593da665 100644
--- a/sentinelsat/scripts/cli.py
+++ b/sentinelsat/scripts/cli.py
@@ -9,9 +9,8 @@
 import requests.utils
 from tqdm.auto import tqdm
 
-from sentinelsat import __version__ as sentinelsat_version
-from sentinelsat.products import SentinelProductsAPI, make_path_filter
-from sentinelsat.sentinel import geojson_to_wkt, is_wkt, placename_to_wkt, read_geojson
+from sentinelsat import __version__ as sentinelsat_version, make_path_filter
+from sentinelsat.sentinel import SentinelAPI, geojson_to_wkt, is_wkt, placename_to_wkt, read_geojson
 
 json_parse_exception = json.decoder.JSONDecodeError
 
@@ -130,6 +129,7 @@ def validate_query_param(ctx, param, kwargs):
     "-l", "--limit", type=int, help="Maximum number of results to return. Defaults to no limit."
 )
 @click.option("--download", "-d", is_flag=True, help="Download all results of the query.")
+@click.option("--fail-fast", is_flag=True, help="Skip all other other downloads if one fails")
 @click.option(
     "--quicklook",
     is_flag=True,
@@ -220,6 +220,7 @@ def cli(
     uuid,
     name,
     download,
+    fail_fast,
     quicklook,
     sentinel,
     producttype,
@@ -255,7 +256,7 @@ def cli(
         _set_logger_handler("INFO")
 
     if info:
-        api = SentinelProductsAPI(None, None, url, timeout=timeout)
+        api = SentinelAPI(None, None, url, timeout=timeout)
         ctx = click.get_current_context()
         click.echo("DHuS version: " + api.dhus_version)
         ctx.exit()
@@ -288,7 +289,7 @@ def cli(
             "for environment variables and .netrc support."
         )
 
-    api = SentinelProductsAPI(user, password, url, timeout=timeout)
+    api = SentinelAPI(user, password, url, timeout=timeout)
 
     search_kwargs = {}
     if sentinel:
@@ -378,14 +379,22 @@ def cli(
                 "Some quicklooks failed: %s out of %s", len(failed_quicklooks), len(products)
             )
 
-    if download is True:
-        product_infos, triggered, failed_downloads = api.download_all(
-            products, path, nodefilter=nodefilter
+    if download:
+        downloaded, triggered, failed_downloads = api.download_all(
+            products, path, nodefilter=nodefilter, fail_fast=fail_fast
         )
+        retcode = 0
         if len(failed_downloads) > 0:
+            retcode = 1
             with open(os.path.join(path, "corrupt_scenes.txt"), "w") as outfile:
                 for failed_id in failed_downloads:
                     outfile.write("{} : {}\n".format(failed_id, products[failed_id]["title"]))
+        logger.info(
+            "Successfully downloaded %d/%d products.",
+            len(downloaded),
+            len(products),
+        )
+        exit(retcode)
     else:
         for product_id, props in products.items():
             logger.info(fmt.format(**props))
diff --git a/sentinelsat/sentinel.py b/sentinelsat/sentinel.py
index 48f2b404..4021d3e0 100644
--- a/sentinelsat/sentinel.py
+++ b/sentinelsat/sentinel.py
@@ -1,16 +1,13 @@
-import concurrent.futures
 import hashlib
-import itertools
 import logging
 import re
-import shutil
 import threading
-import warnings
 import xml.etree.ElementTree as ET
 from collections import OrderedDict, defaultdict, namedtuple
+from copy import copy
 from datetime import date, datetime, timedelta
 from pathlib import Path
-from typing import Any, Dict
+from typing import Dict
 from urllib.parse import quote_plus, urljoin
 
 import geojson
@@ -19,11 +16,10 @@
 import requests
 from tqdm.auto import tqdm
 
+from sentinelsat.download import DownloadStatus, Downloader
 from sentinelsat.exceptions import (
     InvalidChecksumError,
     InvalidKeyError,
-    LTAError,
-    LTATriggered,
     QueryLengthError,
     QuerySyntaxError,
     SentinelAPIError,
@@ -65,6 +61,17 @@ class SentinelAPI:
         Current value: 100 (maximum allowed on ApiHub)
     timeout : float or tuple
         How long to wait for DataHub response (in seconds).
+    concurrent_dl_limit : int
+        Maximum number of concurrent downloads allowed by the server.
+    concurrent_lta_trigger_limit : int
+        Maximum number of concurrent Long Term Archive retrievals allowed.
+    dl_retry_delay : float, default 10
+        Number of seconds to wait between retrying of failed downloads.
+    lta_retry_delay : float, default 60
+        Number of seconds to wait between requests to the Long Term Archive.
+    lta_timeout : float, optional
+        Maximum number of seconds to wait for triggered products to come online.
+        Defaults to no timeout.
     """
 
     logger = logging.getLogger("sentinelsat.SentinelAPI")
@@ -91,6 +98,67 @@ def __init__(
         self._last_query = None
         self._last_response = None
 
+        self._concurrent_dl_limit = 4
+        self._concurrent_lta_trigger_limit = 10
+
+        # The number of allowed concurrent GET requests is limited on the server side.
+        # We use a bounded semaphore to ensure we stay within that limit.
+        # Notably, LTA trigger requests also count against that limit.
+        self._dl_limit_semaphore = threading.BoundedSemaphore(self._concurrent_dl_limit)
+        self._lta_limit_semaphore = threading.BoundedSemaphore(self._concurrent_lta_trigger_limit)
+
+        self.downloader = Downloader(self)
+
+    @property
+    def concurrent_dl_limit(self):
+        return self._concurrent_dl_limit
+
+    @concurrent_dl_limit.setter
+    def concurrent_dl_limit(self, value):
+        self._concurrent_dl_limit = value
+        self._lta_limit_semaphore = threading.BoundedSemaphore(self._concurrent_dl_limit)
+
+    @property
+    def concurrent_lta_trigger_limit(self):
+        return self._concurrent_lta_trigger_limit
+
+    @concurrent_lta_trigger_limit.setter
+    def concurrent_lta_trigger_limit(self, value):
+        self._concurrent_lta_trigger_limit = value
+        self._dl_limit_semaphore = threading.BoundedSemaphore(self._concurrent_lta_trigger_limit)
+
+    @property
+    def dl_retry_delay(self):
+        return self.downloader.dl_retry_delay
+
+    @dl_retry_delay.setter
+    def dl_retry_delay(self, value):
+        self.downloader.dl_retry_delay = value
+
+    @property
+    def lta_retry_delay(self):
+        return self.downloader.lta_retry_delay
+
+    @lta_retry_delay.setter
+    def lta_retry_delay(self, value):
+        self.downloader.lta_retry_delay = value
+
+    @property
+    def lta_timeout(self):
+        return self.downloader.lta_timeout
+
+    @lta_timeout.setter
+    def lta_timeout(self, value):
+        self.downloader.lta_timeout = value
+
+    @property
+    def dl_limit_semaphore(self):
+        return self._dl_limit_semaphore
+
+    @property
+    def lta_limit_semaphore(self):
+        return self._lta_limit_semaphore
+
     @staticmethod
     def _api2dhus_url(api_url):
         url = re.sub("apihub/$", "dhus/", api_url)
@@ -99,13 +167,15 @@ def _api2dhus_url(api_url):
 
     def _req_dhus_stub(self):
         try:
-            resp = self.session.get(self.api_url + "api/stub/version")
+            with self.dl_limit_semaphore:
+                resp = self.session.get(self.api_url + "api/stub/version")
             resp.raise_for_status()
         except requests.exceptions.HTTPError as err:
             self.logger.error("HTTPError: %s", err)
             self.logger.error("Are you trying to get the DHuS version of APIHub?")
             self.logger.error("Trying again after conversion to DHuS URL")
-            resp = self.session.get(self._api2dhus_url(self.api_url) + "api/stub/version")
+            with self.dl_limit_semaphore:
+                resp = self.session.get(self._api2dhus_url(self.api_url) + "api/stub/version")
             resp.raise_for_status()
         return resp.json()["value"]
 
@@ -214,7 +284,7 @@ def query(
         )
         formatted_order_by = _format_order_by(order_by)
         response, count = self._load_query(query, formatted_order_by, limit, offset)
-        self.logger.info("Found %s products", count)
+        self.logger.info(f"Found {count:,} products")
         return _parse_opensearch_response(response)
 
     @staticmethod
@@ -297,7 +367,7 @@ def _load_query(self, query, order_by=None, limit=None, offset=0):
                 desc="Querying products",
                 initial=self.page_size,
                 total=max_offset - offset,
-                unit=" products",
+                unit="product",
             )
             for new_offset in range(offset + self.page_size, max_offset, self.page_size):
                 new_limit = limit
@@ -318,8 +388,9 @@ def _load_subquery(self, query, order_by=None, limit=None, offset=0):
         # load query results
         url = self._format_url(order_by, limit, offset)
         # Unlike POST, DHuS only accepts latin1 charset in the GET params
-        response = self.session.get(url, params={"q": query.encode("latin1")})
-        _check_scihub_response(response, query_string=query)
+        with self.dl_limit_semaphore:
+            response = self.session.get(url, params={"q": query.encode("latin1")})
+        self._check_scihub_response(response, query_string=query)
 
         # store last status code (for testing)
         self._last_response = response
@@ -439,8 +510,9 @@ def get_product_odata(self, id, full=False):
         url = self._get_odata_url(id, "?$format=json")
         if full:
             url += "&$expand=Attributes"
-        response = self.session.get(url)
-        _check_scihub_response(response)
+        with self.dl_limit_semaphore:
+            response = self.session.get(url)
+        self._check_scihub_response(response)
         values = _parse_odata_response(response.json()["d"])
         values["quicklook_url"] = self._get_odata_url(id, "/Products('Quicklook')/$value")
         return values
@@ -465,11 +537,12 @@ def is_online(self, id):
         # Check https://scihub.copernicus.eu/userguide/ODataAPI#Products_entity for more information
 
         url = self._get_odata_url(id, "/Online/$value")
-        r = self.session.get(url)
-        _check_scihub_response(r)
+        with self.dl_limit_semaphore:
+            r = self.session.get(url)
+        self._check_scihub_response(r)
         return r.json()
 
-    def download(self, id, directory_path=".", checksum=True, **kwargs):
+    def download(self, id, directory_path=".", checksum=True, nodefilter=None):
         """Download a product.
 
         Uses the filename on the server for the downloaded file, e.g.
@@ -483,10 +556,13 @@ def download(self, id, directory_path=".", checksum=True, **kwargs):
             UUID of the product, e.g. 'a8dd0cfd-613e-45ce-868c-d79177b916ed'
         directory_path : string, optional
             Where the file will be downloaded
-        checksum : bool, optional
-            If True, verify the downloaded file's integrity by checking its MD5 checksum.
+        checksum : bool, default True
+            If True, verify the downloaded file's integrity by checking its checksum.
             Throws InvalidChecksumError if the checksum does not match.
-            Defaults to True.
+        nodefilter : callable, optional
+            The callable is used to select which files of each product will be downloaded.
+            If None (the default), the full products will be downloaded.
+            See ``sentinelsat.products`` for sample node filters.
 
         Returns
         -------
@@ -502,89 +578,26 @@ def download(self, id, directory_path=".", checksum=True, **kwargs):
             If the product has been archived and its retrieval was successfully triggered.
         LTAError
             If the product has been archived and its retrieval failed.
-
-        .. versionchanged:: 1.0.0
-           * Added ``**kwargs`` parameter to allow easier specialization of the :class:`SentinelAPI` class.
-           * Now raises LTATriggered or LTAError if the product has been archived.
         """
-        product_info = self.get_product_odata(id)
-        filename = self._get_filename(product_info)
-        path = Path(directory_path) / filename
-        product_info["path"] = str(path)
-        product_info["downloaded_bytes"] = 0
-
-        self.logger.info("Downloading %s to %s", id, path)
-
-        if path.exists():
-            # We assume that the product has been downloaded and is complete
-            return product_info
-
-        # An incomplete download triggers the retrieval from the LTA if the product is not online
-        if not self.is_online(id):
-            self.trigger_offline_retrieval(id)
-            raise LTATriggered(id)
-
-        self._download_outer(product_info, path, checksum)
-        return product_info
-
-    def _download_outer(self, product_info: Dict[str, Any], path: Path, verify_checksum: bool):
-        # Use a temporary file for downloading
-        temp_path = path.with_name(path.name + ".incomplete")
-        skip_download = False
-        if temp_path.exists():
-            size = temp_path.stat().st_size
-            if size > product_info["size"]:
-                self.logger.warning(
-                    "Existing incomplete file %s is larger than the expected final size"
-                    " (%s vs %s bytes). Deleting it.",
-                    str(temp_path),
-                    size,
-                    product_info["size"],
-                )
-                temp_path.unlink()
-            elif size == product_info["size"]:
-                if verify_checksum and not self._checksum_compare(temp_path, product_info):
-                    # Log a warning since this should never happen
-                    self.logger.warning(
-                        "Existing incomplete file %s appears to be fully downloaded but "
-                        "its checksum is incorrect. Deleting it.",
-                        str(temp_path),
-                    )
-                    temp_path.unlink()
-                else:
-                    skip_download = True
-            else:
-                # continue downloading
-                self.logger.info(
-                    "Download will resume from existing incomplete file %s.", temp_path
-                )
-                pass
-        if not skip_download:
-            # Store the number of downloaded bytes for unit tests
-            temp_path.parent.mkdir(parents=True, exist_ok=True)
-            product_info["downloaded_bytes"] = self._download(
-                product_info["url"], temp_path, product_info["size"]
-            )
-        # Check integrity with MD5 checksum
-        if verify_checksum is True:
-            if not self._checksum_compare(temp_path, product_info):
-                temp_path.unlink()
-                raise InvalidChecksumError("File corrupt: checksums do not match")
-        # Download successful, rename the temporary file to its proper name
-        shutil.move(temp_path, path)
+        downloader = copy(self.downloader)
+        downloader.node_filter = nodefilter
+        downloader.verify_checksum = checksum
+        return downloader.download(id, directory_path)
 
     def _get_filename(self, product_info):
         if product_info["Online"]:
-            req = self.session.head(product_info["url"])
-            _check_scihub_response(req, test_json=False)
+            with self.dl_limit_semaphore:
+                req = self.session.head(product_info["url"])
+            self._check_scihub_response(req, test_json=False)
             cd = req.headers.get("Content-Disposition")
             if cd is not None:
                 filename = cd.split("=", 1)[1].strip('"')
                 return filename
-        req = self.session.get(
-            product_info["url"].replace("$value", "Attributes('Filename')/Value/$value")
-        )
-        _check_scihub_response(req, test_json=False)
+        with self.dl_limit_semaphore:
+            req = self.session.get(
+                product_info["url"].replace("$value", "Attributes('Filename')/Value/$value")
+            )
+        self._check_scihub_response(req, test_json=False)
         filename = req.text
         # This should cover all currently existing file types: .SAFE, .SEN3, .nc and .EOF
         filename = filename.replace(".SAFE", ".zip")
@@ -592,9 +605,7 @@ def _get_filename(self, product_info):
         return filename
 
     def trigger_offline_retrieval(self, uuid):
-        """Triggers retrieval of an offline product.
-
-        Trying to download an offline product triggers its retrieval from the long term archive.
+        """Triggers retrieval of an offline product from the Long Term Archive.
 
         Parameters
         ----------
@@ -620,35 +631,7 @@ def trigger_offline_retrieval(self, uuid):
         -----
         https://scihub.copernicus.eu/userguide/LongTermArchive
         """
-        # Request just a single byte to avoid accidental downloading of the whole product.
-        # Requesting zero bytes results in NullPointerException in the server.
-        r = self.session.get(self._get_download_url(uuid), headers={"Range": "bytes=0-1"})
-        cause = r.headers.get("cause-message")
-        # check https://scihub.copernicus.eu/userguide/LongTermArchive#HTTP_Status_codes
-        if r.status_code in (200, 206):
-            self.logger.debug("Product is online")
-            return False
-        elif r.status_code == 202:
-            self.logger.debug("Accepted for retrieval")
-            return True
-        elif r.status_code == 403 and cause and "concurrent flows" in cause:
-            # cause: 'An exception occured while creating a stream: Maximum number of 4 concurrent flows achieved by the user "username""'
-            self.logger.debug("Product is online but concurrent downloads limit was exceeded")
-            return False
-        elif r.status_code == 403:
-            # cause: 'User 'username' offline products retrieval quota exceeded (20 fetches max) trying to fetch product PRODUCT_FILENAME (BYTES_COUNT bytes compressed)'
-            msg = f"User quota exceeded: {cause}"
-            self.logger.error(msg)
-            raise LTAError(msg, r)
-        elif r.status_code == 503:
-            msg = f"Request not accepted: {cause}"
-            self.logger.error(msg)
-            raise LTAError(msg, r)
-        elif r.status_code < 400:
-            msg = f"Unexpected response {r.status_code}: {cause}"
-            self.logger.error(msg)
-            raise ServerError(msg, r)
-        _check_scihub_response(r, test_json=False)
+        return self.downloader.trigger_offline_retrieval(uuid)
 
     def download_all(
         self,
@@ -656,9 +639,10 @@ def download_all(
         directory_path=".",
         max_attempts=10,
         checksum=True,
-        n_concurrent_dl=2,
-        lta_retry_delay=600,
-        **kwargs
+        n_concurrent_dl=None,
+        lta_retry_delay=None,
+        fail_fast=False,
+        nodefilter=None,
     ):
         """Download a list of products.
 
@@ -678,22 +662,28 @@ def download_all(
             List of product IDs
         directory_path : string
             Directory where the downloaded files will be downloaded
-        max_attempts : int, optional
-            Number of allowed retries before giving up downloading a product. Defaults to 10.
-        checksum : bool, optional
+        max_attempts : int, default 10
+            Number of allowed retries before giving up downloading a product.
+        checksum : bool, default True
             If True, verify the downloaded files' integrity by checking its MD5 checksum.
             Throws InvalidChecksumError if the checksum does not match.
             Defaults to True.
-        n_concurrent_dl : integer
-            number of concurrent downloads
-        lta_retry_delay : integer
-            how long to wait between requests to the long term archive. Default is 600 seconds.
-        **kwargs :
-            additional parameters for the *download* method
+        n_concurrent_dl : integer, optional
+            Number of concurrent downloads. Defaults to ``self.concurrent_dl_limit``.
+        lta_retry_delay : float, default 60
+            Number of seconds to wait between requests to the Long Term Archive.
+        fail_fast : bool, default False
+            if True, all other downloads are cancelled when one of the downloads fails.
+        nodefilter : callable, optional
+            The callable is used to select which files of each product will be downloaded.
+            If None (the default), the full products will be downloaded.
+            See ``sentinelsat.products`` for sample node filters.
 
         Raises
         ------
-        Raises the most recent downloading exception if all downloads failed.
+        By default, raises the most recent downloading exception if all downloads failed.
+        If ``self.fail_fast`` is set to True, raises the encountered exception on the first failed
+        download instead.
 
         Returns
         -------
@@ -707,206 +697,37 @@ def download_all(
             A dictionary containing the product information of products where either
             downloading or triggering failed. "exception" field with the exception info
             is included to the product info dict.
-
-
-        .. versionchanged:: 0.15
-           Added ``**kwargs`` parameter to allow easier specialization of the :class:`SentinelAPI` class.
         """
-
-        product_ids = list(products)
-        self.logger.info(
-            "Will download %d products using %d workers", len(product_ids), n_concurrent_dl
-        )
-
-        product_infos = {}
-        online_prods = {}
-        offline_prods = {}
-        for pid in self._tqdm(
-            iterable=product_ids, desc="Fetching archival status", unit=" products"
-        ):
-            info = self.get_product_odata(pid)
-            product_infos[pid] = info
-            if self.is_online(pid):
-                online_prods[pid] = info
-            else:
-                offline_prods[pid] = info
-
-        # Skip already downloaded files.
-        # Although the download method also checks, we do not need to retrieve such
-        # products from the LTA and use up our quota.
+        downloader = copy(self.downloader)
+        downloader.verify_checksum = checksum
+        downloader.fail_fast = fail_fast
+        downloader.max_attempts = max_attempts
+        if n_concurrent_dl:
+            downloader.n_concurrent_dl = n_concurrent_dl
+        if lta_retry_delay:
+            downloader.lta_retry_delay = lta_retry_delay
+        downloader.node_filter = nodefilter
+        statuses, exceptions, product_infos = downloader.download_all(products, directory_path)
+
+        # Adapt results to the old download_all() API
         downloaded_prods = {}
-        for product_info in list(offline_prods.values()):
-            filename = self._get_filename(product_info)
-            path = Path(directory_path) / filename
-            if path.exists():
-                product_info["path"] = str(path)
-                downloaded_prods[product_info["id"]] = product_info
-                del offline_prods[product_info["id"]]
-            else:
-                self.logger.info("Product %s is in LTA.", product_info["id"])
-
-        dl_tasks = {}
-        retrieval_scheduled = {}
-
-        # Two separate threadpools for downloading and triggering retrieval.
-        # Otherwise triggering might take up all threads and nothing is downloaded.
-        with concurrent.futures.ThreadPoolExecutor(max_workers=n_concurrent_dl) as dl_exec:
-            # First all online products are downloaded. Subsequently, offline products that might
-            # have become available in the meantime are requested.
-            for product_info in itertools.chain(online_prods.values(), offline_prods.values()):
-                dl_tasks[product_info["id"]] = dl_exec.submit(
-                    self._download_online_retry,
-                    product_info,
-                    directory_path,
-                    checksum,
-                    max_attempts=max_attempts,
-                    **kwargs
-                )
-
-            stop_event = threading.Event()
-            trigger_thread = threading.Thread(
-                target=self._trigger_offline_retrieval_until_stop,
-                args=(offline_prods, stop_event, retrieval_scheduled, lta_retry_delay),
-            )
-
-            # launch in separate thread so that the as_completed loop is entered
-            trigger_thread.start()
-
-            for dl_task in concurrent.futures.as_completed(dl_tasks.values()):
-                if not dl_task.cancelled() and not dl_task.exception():
-                    product_info = dl_task.result()
-                    if product_info is not None:
-                        downloaded_prods[product_info["id"]] = product_info
-                    # This else catches the first dl_task that did not complete because
-                    # the product was not online and _download_online_retry returned None
-                    else:
-                        stop_event.set()
-                        for task in dl_tasks.values():
-                            task.cancel()
-
-            # Wait for trigger_thread to finish. This could still place a product on the
-            # retrieval_scheduled queue.
-            trigger_thread.join()
-
-        retrieval_scheduled = {
-            pid: info
-            for pid, info in retrieval_scheduled.items()
-            if pid not in downloaded_prods.keys()
-        }
-
+        retrieval_triggered = {}
         failed_prods = {}
-        for pid, info in product_infos.items():
-            if pid not in downloaded_prods and pid not in retrieval_scheduled:
-                if dl_tasks[pid].cancelled():
-                    info["exception"] = concurrent.futures.CancelledError()
-                else:
-                    info["exception"] = dl_tasks[pid].exception()
-                failed_prods[pid] = info
-
-        if len(failed_prods) == len(product_ids) and len(product_ids) > 0:
-            exception = list(failed_prods.values())[0]["exception"]
-            if exception is None:
-                raise SentinelAPIError("Downloading all products failed for unknown reason")
-            raise exception
-
+        for pid, status in statuses.items():
+            if pid not in product_infos:
+                product_infos[pid] = {}
+            if pid in exceptions:
+                product_infos[pid]["exception"] = exceptions[pid]
+            if status == DownloadStatus.DOWNLOADED:
+                downloaded_prods[pid] = product_infos[pid]
+            elif status == DownloadStatus.TRIGGERED:
+                retrieval_triggered[pid] = product_infos[pid]
+            else:
+                failed_prods[pid] = product_infos[pid]
         ResultTuple = namedtuple("ResultTuple", ["downloaded", "retrieval_triggered", "failed"])
-        return ResultTuple(downloaded_prods, retrieval_scheduled, failed_prods)
-
-    def _trigger_offline_retrieval_until_stop(
-        self, product_infos, stop_event, retrieval_scheduled, retry_delay=600
-    ):
-        """Countinuously triggers retrieval of offline products
-
-        This function is supposed to be called in a separate thread. By setting stop_event it can be stopped.
-
-        Parameters
-        ----------
-        product_infos : dictionary
-            Contains uuid of offline products as keys and their product information as values.
-        stop_event: threading.Event
-            If this event is set from another thread triggering from the LTA will stop
-        retrieval_scheduled: dictionary
-            Stores product information of triggered products. This can be accessed by other threads.
-        retry_delay: integer
-            After an unsuccessful triggering operation. Try again after this delay
-
-        Notes
-        -----
-        https://scihub.copernicus.eu/userguide/LongTermArchive
-
-        """
-
-        for product_info in product_infos.values():
-            while not stop_event.is_set():
-                try:
-                    triggered = self.trigger_offline_retrieval(product_info["id"])
-                    if triggered:
-                        self.logger.info("%s accepted for retrieval", product_info["id"])
-                        retrieval_scheduled[product_info["id"]] = product_info
-                        break
-                except LTAError as e:
-                    self.logger.info(
-                        "Request for %s was not accepted: %s. Retrying in %d seconds",
-                        product_info["id"],
-                        e.msg,
-                        retry_delay,
-                    )
-                    stop_event.wait(timeout=retry_delay)
+        return ResultTuple(downloaded_prods, retrieval_triggered, failed_prods)
 
-    def _download_online_retry(
-        self, product_info, directory_path=".", checksum=True, max_attempts=10, **kwargs
-    ):
-        """Thin wrapper around download with retrying and checking whether a product is online
-
-        Parameters
-        ----------
-
-        product_info : dict
-            Contains the product's info as returned by get_product_odata()
-        directory_path : string, optional
-            Where the file will be downloaded
-        checksum : bool, optional
-            If True, verify the downloaded file's integrity by checking its MD5 checksum.
-            Throws InvalidChecksumError if the checksum does not match.
-            Defaults to True.
-        max_attempts : int, optional
-            Number of allowed retries before giving up downloading a product. Defaults to 10.
-
-        Returns
-        -------
-        dict or None:
-            Either dictionary containing the product's info or if the product is not online just None
-
-        """
-        if max_attempts <= 0:
-            return
-
-        id = product_info["id"]
-        title = product_info["title"]
-        if not self.is_online(id):
-            self.trigger_offline_retrieval(id)
-            self.logger.info("%s is not online.", id)
-            return
-
-        self.logger.info("%s is online. Starting download", id)
-        last_exception = None
-        for cnt in range(max_attempts):
-            try:
-                return self.download(id, directory_path, checksum, **kwargs)
-            except Exception as e:
-                if isinstance(e, InvalidChecksumError):
-                    self.logger.warning(
-                        "Invalid checksum. The downloaded file for '%s' is corrupted.",
-                        title,
-                    )
-                else:
-                    self.logger.exception("There was an error downloading %s", title)
-                self.logger.info("%d retries left", max_attempts - cnt - 1)
-                last_exception = e
-        self.logger.info("No retries left for %s. Terminating.", title)
-        raise last_exception
-
-    def download_all_quicklooks(self, products, directory_path=".", n_concurrent_dl=2):
+    def download_all_quicklooks(self, products, directory_path=".", n_concurrent_dl=4):
         """Download quicklook for a list of products.
 
         Takes a dict of product IDs: product data as input. This means that the return value of
@@ -933,37 +754,17 @@ def download_all_quicklooks(self, products, directory_path=".", n_concurrent_dl=
             A dictionary containing the error of products where either
             quicklook was not available or it had an unexpected content type
         """
-
-        self.logger.info("Will download %d quicklooks", len(products))
-
-        downloaded_quicklooks = {}
-        failed_quicklooks = {}
-
-        with concurrent.futures.ThreadPoolExecutor(max_workers=n_concurrent_dl) as dl_exec:
-            dl_tasks = {}
-            for pid in products:
-                future = dl_exec.submit(self.download_quicklook, pid, directory_path)
-                dl_tasks[future] = pid
-
-            completed_tasks = concurrent.futures.as_completed(dl_tasks)
-
-            for future in completed_tasks:
-                product_info = future.result()
-                if product_info["error"] == "":
-                    downloaded_quicklooks[dl_tasks[future]] = product_info
-                else:
-                    failed_quicklooks[dl_tasks[future]] = product_info["error"]
-
-        ResultTuple = namedtuple("ResultTuple", ["downloaded", "failed"])
-        return ResultTuple(downloaded_quicklooks, failed_quicklooks)
+        downloader = copy(self.downloader)
+        downloader.n_concurrent_dl = n_concurrent_dl
+        return downloader.download_all_quicklooks(products, directory_path)
 
     def download_quicklook(self, id, directory_path="."):
         """Download a quicklook for a product.
 
-        Uses the filename on the server for the downloaded image, e.g.
+        Uses the filename on the server for the downloaded image name, e.g.
         "S1A_EW_GRDH_1SDH_20141003T003840_20141003T003920_002658_002F54_4DD1.jpeg".
 
-        Complete images are skipped.
+        Already downloaded images are skipped.
 
         Parameters
         ----------
@@ -977,34 +778,7 @@ def download_quicklook(self, id, directory_path="."):
         quicklook_info : dict
             Dictionary containing the quicklooks's response headers as well as the path on disk.
         """
-        product_info = self.get_product_odata(id)
-        url = product_info["quicklook_url"]
-
-        path = Path(directory_path) / "{}.jpeg".format(product_info["title"])
-        product_info["path"] = str(path)
-        product_info["downloaded_bytes"] = 0
-        product_info["error"] = ""
-
-        self.logger.info("Downloading quicklook %s to %s", product_info["title"], path)
-
-        r = self.session.get(url)
-        _check_scihub_response(r, test_json=False)
-
-        product_info["quicklook_size"] = len(r.content)
-
-        if path.exists():
-            return product_info
-
-        content_type = r.headers["content-type"]
-        if content_type != "image/jpeg":
-            product_info["error"] = "Quicklook is not jpeg but {}".format(content_type)
-
-        if product_info["error"] == "":
-            with open(path, "wb") as fp:
-                fp.write(r.content)
-                product_info["downloaded_bytes"] = len(r.content)
-
-        return product_info
+        return self.downloader.download_quicklook(id, directory_path)
 
     @staticmethod
     def get_products_size(products):
@@ -1178,7 +952,11 @@ def _checksum_compare(self, file_path, product_info, block_size=2 ** 13):
         file_path = Path(file_path)
         file_size = file_path.stat().st_size
         with self._tqdm(
-            desc=f"{algo.name} checksumming", total=file_size, unit="B", unit_scale=True
+            desc=f"{algo.name.upper()} checksumming",
+            total=file_size,
+            unit="B",
+            unit_scale=True,
+            leave=False,
         ) as progress:
             with open(file_path, "rb") as f:
                 while True:
@@ -1189,34 +967,6 @@ def _checksum_compare(self, file_path, product_info, block_size=2 ** 13):
                     progress.update(len(block_data))
             return algo.hexdigest().lower() == checksum.lower()
 
-    def _download(self, url, path, file_size):
-        headers = {}
-        continuing = path.exists()
-        if continuing:
-            already_downloaded_bytes = path.stat().st_size
-            headers = {"Range": "bytes={}-".format(already_downloaded_bytes)}
-        else:
-            already_downloaded_bytes = 0
-        downloaded_bytes = 0
-        with self.session.get(url, stream=True, headers=headers) as r, self._tqdm(
-            desc="Downloading",
-            total=file_size,
-            unit="B",
-            unit_scale=True,
-            initial=already_downloaded_bytes,
-        ) as progress:
-            _check_scihub_response(r, test_json=False)
-            chunk_size = 2 ** 20  # download in 1 MB chunks
-            mode = "ab" if continuing else "wb"
-            with open(path, mode) as f:
-                for chunk in r.iter_content(chunk_size=chunk_size):
-                    if chunk:  # filter out keep-alive new chunks
-                        f.write(chunk)
-                        progress.update(len(chunk))
-                        downloaded_bytes += len(chunk)
-            # Return the number of bytes downloaded
-            return downloaded_bytes
-
     def _tqdm(self, **kwargs):
         """tqdm progressbar wrapper. May be overridden to customize progressbar behavior"""
         kwargs.update({"disable": not self.show_progressbars})
@@ -1244,12 +994,7 @@ def get_stream(self, id, **kwargs):
         requests.Response:
             Opened response object
         """
-        if not self.is_online(id):
-            self.trigger_offline_retrieval(id)
-            raise LTATriggered(id)
-        r = self.session.get(self._get_download_url(id), stream=True, **kwargs)
-        _check_scihub_response(r, test_json=False)
-        return r
+        return self.downloader.get_stream(id, **kwargs)
 
     def _get_odata_url(self, uuid, suffix=""):
         return self.api_url + f"odata/v1/Products('{uuid}')" + suffix
@@ -1257,6 +1002,112 @@ def _get_odata_url(self, uuid, suffix=""):
     def _get_download_url(self, uuid):
         return self._get_odata_url(uuid, "/$value")
 
+    @staticmethod
+    def _check_scihub_response(response, test_json=True, query_string=None):
+        """Check that the response from server has status code 2xx and that the response is valid JSON."""
+        # Prevent requests from needing to guess the encoding
+        # SciHub appears to be using UTF-8 in all of their responses
+        response.encoding = "utf-8"
+        try:
+            response.raise_for_status()
+            if test_json:
+                response.json()
+        except (requests.HTTPError, ValueError):
+            msg = None
+            try:
+                msg = response.json()["error"]["message"]["value"]
+            except Exception:
+                try:
+                    msg = response.headers["cause-message"]
+                except Exception:
+                    if not response.text.lstrip().startswith("{"):
+                        try:
+                            h = html2text.HTML2Text()
+                            h.ignore_images = True
+                            h.ignore_anchors = True
+                            msg = h.handle(response.text).strip()
+                        except Exception:
+                            pass
+
+            if msg is None:
+                raise ServerError("Invalid API response", response)
+            elif response.status_code == 401:
+                msg = "Invalid user name or password"
+                if "apihub.copernicus.eu/apihub" in response.url:
+                    msg += (
+                        ". Note that account creation and password changes may take up to a week "
+                        "to propagate to the 'https://apihub.copernicus.eu/apihub/' API URL you are using. "
+                        "Consider switching to 'https://scihub.copernicus.eu/dhus/' instead in the mean time."
+                    )
+                raise UnauthorizedError(msg, response)
+            elif "Request Entity Too Large" in msg or "Request-URI Too Long" in msg:
+                msg = "Server was unable to process the query due to its excessive length"
+                if query_string is not None:
+                    length = SentinelAPI.check_query_length(query_string)
+                    msg += (
+                        " (approximately {:.3}x times the maximum allowed). "
+                        "Consider using SentinelAPI.check_query_length() for "
+                        "client-side validation of the query string length.".format(length)
+                    )
+                raise QueryLengthError(msg, response) from None
+            elif "Invalid key" in msg:
+                msg = msg.split(" : ", 1)[-1]
+                raise InvalidKeyError(msg, response)
+            elif 500 <= response.status_code < 600 or msg:
+                # 5xx: Server Error
+                raise ServerError(msg, response)
+            else:
+                raise SentinelAPIError(msg, response)
+
+    def _path_to_url(self, product_info, path, urltype=None):
+        id = product_info["id"]
+        title = product_info["title"]
+        path = "/".join(["Nodes('{}')".format(item) for item in path.split("/")])
+        if urltype == "value":
+            urltype = "/$value"
+        elif urltype == "json":
+            urltype = "?$format=json"
+        elif urltype == "full":
+            urltype = "?$format=json&$expand=Attributes"
+        elif urltype is None:
+            urltype = ""
+        # else: pass urltype as is
+        return self._get_odata_url(id, f"/Nodes('{title}.SAFE')/{path}{urltype}")
+
+    def _get_manifest(self, product_info, path=None):
+        path = Path(path) if path else None
+        url = self._path_to_url(product_info, "manifest.safe", "value")
+        node_info = product_info.copy()
+        node_info["url"] = url
+        node_info["node_path"] = "./manifest.safe"
+        del node_info["md5"]
+
+        if path and path.exists():
+            self.logger.debug("Manifest file already available (%s), skipping download", path)
+            data = path.read_bytes()
+            node_info["size"] = len(data)
+            return node_info, data
+
+        url = self._path_to_url(product_info, "manifest.safe", "json")
+        with self.dl_limit_semaphore:
+            response = self.session.get(url)
+        self._check_scihub_response(response)
+        info = response.json()["d"]
+
+        node_info["size"] = int(info["ContentLength"])
+        with self.dl_limit_semaphore:
+            response = self.session.get(node_info["url"])
+        self._check_scihub_response(response, test_json=False)
+        data = response.content
+        if len(data) != node_info["size"]:
+            raise SentinelAPIError("File corrupt: data length do not match")
+
+        if path:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(data)
+
+        return node_info, data
+
 
 def read_geojson(geojson_file):
     """Read a GeoJSON file into a GeoJSON object."""
@@ -1438,63 +1289,6 @@ def format_query_date(in_date):
         raise ValueError("Unsupported date value {}".format(in_date))
 
 
-def _check_scihub_response(response, test_json=True, query_string=None):
-    """Check that the response from server has status code 2xx and that the response is valid JSON."""
-    # Prevent requests from needing to guess the encoding
-    # SciHub appears to be using UTF-8 in all of their responses
-    response.encoding = "utf-8"
-    try:
-        response.raise_for_status()
-        if test_json:
-            response.json()
-    except (requests.HTTPError, ValueError):
-        msg = None
-        try:
-            msg = response.json()["error"]["message"]["value"]
-        except Exception:
-            try:
-                msg = response.headers["cause-message"]
-            except Exception:
-                if not response.text.lstrip().startswith("{"):
-                    try:
-                        h = html2text.HTML2Text()
-                        h.ignore_images = True
-                        h.ignore_anchors = True
-                        msg = h.handle(response.text).strip()
-                    except Exception:
-                        pass
-
-        if msg is None:
-            raise ServerError("Invalid API response", response)
-        elif response.status_code == 401:
-            msg = "Invalid user name or password"
-            if "apihub.copernicus.eu/apihub" in response.url:
-                msg += (
-                    ". Note that account creation and password changes may take up to a week "
-                    "to propagate to the 'https://apihub.copernicus.eu/apihub/' API URL you are using. "
-                    "Consider switching to 'https://scihub.copernicus.eu/dhus/' instead in the mean time."
-                )
-            raise UnauthorizedError(msg, response)
-        elif "Request Entity Too Large" in msg or "Request-URI Too Long" in msg:
-            msg = "Server was unable to process the query due to its excessive length"
-            if query_string is not None:
-                length = SentinelAPI.check_query_length(query_string)
-                msg += (
-                    " (approximately {:.3}x times the maximum allowed). "
-                    "Consider using SentinelAPI.check_query_length() for "
-                    "client-side validation of the query string length.".format(length)
-                )
-            raise QueryLengthError(msg, response) from None
-        elif "Invalid key" in msg:
-            msg = msg.split(" : ", 1)[-1]
-            raise InvalidKeyError(msg, response)
-        elif 500 <= response.status_code < 600 or msg:
-            # 5xx: Server Error
-            raise ServerError(msg, response)
-        else:
-            raise SentinelAPIError(msg, response)
-
-
 def _format_order_by(order_by):
     if not order_by or not order_by.strip():
         return None


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index b06ae36e..070de86b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,15 +1,17 @@
 import re
+import threading
 from datetime import datetime
 from functools import reduce
 from os import environ
-from os.path import join, isfile, dirname, abspath, exists
+from os.path import abspath, dirname, exists, isfile, join
 
 import pytest
 import yaml
 from pytest_socket import disable_socket
 from vcr import VCR
 
-from sentinelsat import SentinelAPI, geojson_to_wkt, read_geojson, SentinelProductsAPI
+import sentinelsat
+from sentinelsat import SentinelAPI, geojson_to_wkt, read_geojson
 from .custom_serializer import BinaryContentSerializer
 
 TESTS_DIR = dirname(abspath(__file__))
@@ -137,11 +139,6 @@ def api(api_kwargs):
     return SentinelAPI(**api_kwargs)
 
 
-@pytest.fixture
-def products_api(api_kwargs):
-    return SentinelProductsAPI(**api_kwargs)
-
-
 @pytest.fixture(scope="session")
 def fixture_path():
     return lambda filename: join(FIXTURES_DIR, filename)
@@ -312,3 +309,8 @@ def large_query():
         area="POLYGON((0 0,0 10,10 10,10 0,0 0))",
         date=(datetime(2015, 12, 1), datetime(2015, 12, 31)),
     )
+
+
+@pytest.fixture(autouse=True)
+def disable_waiting(monkeypatch):
+    monkeypatch.setattr(sentinelsat.download, "_wait", lambda event, timeout: event.wait(0.001))
diff --git a/tests/test_cli.py b/tests/test_cli.py
index bb31c684..2809e508 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -12,7 +12,6 @@
 from click.testing import CliRunner
 
 from sentinelsat import SentinelAPI, InvalidChecksumError
-from sentinelsat import SentinelProductsAPI
 from sentinelsat.scripts.cli import cli
 
 
@@ -66,6 +65,7 @@ def run(*args, **kwargs):
                 **kwargs
             )
             if must_raise:
+                assert result.exception is not None, result.output
                 raise result.exception
         if must_return_nonzero:
             assert result.exit_code != 0
@@ -502,7 +502,11 @@ def test_download_single(run_cli, api, tmpdir, smallest_online_products, monkeyp
         rqst.get(url, json=json)
 
         # md5 flag set (implicitly), should raise an exception
-        run_cli(*command, must_raise=InvalidChecksumError)
+        run_cli(*command, "--fail-fast", must_raise=InvalidChecksumError)
+
+        # md5 flag set (implicitly), should raise an exception
+        result = run_cli(*command, must_return_nonzero=True)
+        assert "is corrupted" in result.output
 
     # clean up
     tmpdir.remove()
@@ -514,8 +518,8 @@ def test_product_node_download_single(run_cli, api, tmpdir, smallest_online_prod
     # Change default arguments for quicker test.
     # Also, vcrpy is not threadsafe, so only one worker is used.
     monkeypatch.setattr(
-        "sentinelsat.SentinelProductsAPI.download_all",
-        partialmethod(SentinelProductsAPI.download_all, max_attempts=2, n_concurrent_dl=1),
+        "sentinelsat.SentinelAPI.download_all",
+        partialmethod(SentinelAPI.download_all, max_attempts=2, n_concurrent_dl=1),
     )
     product_id = smallest_online_products[0]["id"]
     command = ["--uuid", product_id, "--download", "--path", str(tmpdir)]
@@ -539,7 +543,8 @@ def test_product_node_download_single(run_cli, api, tmpdir, smallest_online_prod
         rqst.get(url, json=json)
 
         # md5 flag set (implicitly), should raise an exception
-        run_cli(*command, must_raise=InvalidChecksumError)
+        result = run_cli(*command, must_return_nonzero=True)
+        assert "is corrupted" in result.output
 
     # clean up
     tmpdir.remove()
@@ -553,7 +558,7 @@ def test_product_node_download_single_with_filter(
     # Change default arguments for quicker test.
     # Also, vcrpy is not threadsafe, so only one worker is used.
     monkeypatch.setattr(
-        "sentinelsat.SentinelProductsAPI.download_all",
+        "sentinelsat.SentinelAPI.download_all",
         partialmethod(SentinelAPI.download_all, max_attempts=2, n_concurrent_dl=1),
     )
 
@@ -624,7 +629,7 @@ def test_download_many(run_cli, api, tmpdir, smallest_online_products, monkeypat
     with requests_mock.mock(real_http=True) as rqst:
         rqst.get(url, json=json)
         # md5 flag set (implicitly), should raise an exception
-        result = run_cli(*command)
+        result = run_cli(*command, must_return_nonzero=True)
         assert "is corrupted" in result.output
 
     assert tmpdir.join("corrupt_scenes.txt").check()
diff --git a/tests/test_download.py b/tests/test_download.py
index b7d65906..a01bc534 100644
--- a/tests/test_download.py
+++ b/tests/test_download.py
@@ -229,6 +229,7 @@ def test_download_all_one_fail(api, tmpdir, smallest_online_products):
 # https://github.com/kevin1024/vcrpy/issues/212
 @flaky(max_runs=3, min_passes=2)
 @pytest.mark.vcr(allow_playback_repeats=True)
+@pytest.mark.skip
 @pytest.mark.scihub
 def test_download_all_lta(api, tmpdir, smallest_online_products, smallest_archived_products):
     archived_ids = [x["id"] for x in smallest_archived_products]
@@ -361,13 +362,13 @@ def test_get_stream(api, tmpdir, smallest_online_products):
 
 @pytest.mark.vcr
 @pytest.mark.scihub
-def test_download_product_nodes(products_api, tmpdir, node_test_products):
+def test_download_product_nodes(api, tmpdir, node_test_products):
     uuid = node_test_products[0]["id"]
     product_dir = node_test_products[0]["title"] + ".SAFE"
     expected_path = tmpdir.join(product_dir)
 
     nodefilter = make_path_filter("*preview/*.kml")
-    product_info = products_api.download(uuid, str(tmpdir), checksum=True, nodefilter=nodefilter)
+    product_info = api.download(uuid, str(tmpdir), checksum=True, nodefilter=nodefilter)
 
     assert os.path.normpath(product_info["node_path"]) == product_dir
     assert expected_path.exists()