Source code for

Module implementing a download client that is adjusted to download from AWS

import logging
import warnings
from typing import Any, Dict, Optional
from typing_extensions import deprecated

    from boto3 import Session
    from botocore.exceptions import NoCredentialsError
except ImportError as import_exception:
    raise ImportError(
        "To use AWS functionalities of this package you have to install sentinelhub[AWS] package extension"
    ) from import_exception

from ..config import SHConfig
from import DownloadClient
from import fail_missing_file
from import DownloadRequest, DownloadResponse
from ..exceptions import AwsDownloadFailedException, SHDeprecationWarning

LOGGER = logging.getLogger(__name__)

[docs]@deprecated( "AWS functionality will remain in the codebase for now, but won't be actively maintained.", category=SHDeprecationWarning, ) class AwsDownloadClient(DownloadClient): """An AWS download client class""" GLOBAL_S3_CLIENTS: Dict[str, Any] = {} def __init__(self, *args: Any, boto_params: Optional[Dict[str, Any]] = None, **kwargs: Any): """ :param args: Positional arguments propagated to `DownloadClient` class. :param boto_params: A dictionary of extra parameters that will be propagated to `botocore.client.S3.get_object` method. E.g. `{"RequestPayer": "requester"}`. :param kwargs: Keyword arguments propagated to `DownloadClient` class. """ super().__init__(*args, **kwargs) self.boto_params = boto_params or {} @fail_missing_file def _execute_download(self, request: DownloadRequest) -> DownloadResponse: """Executes a download procedure""" if not self.is_s3_request(request): return super()._execute_download(request) s3_client = self.get_s3_client(self.config) response_content = self._do_download(request, s3_client) LOGGER.debug("Successful download from %s", request.url) return DownloadResponse(request=request, content=response_content)
[docs] @classmethod def get_s3_client(cls, config: SHConfig) -> Any: """Provides a s3 client object""" warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<ssl.SSLSocket.*>") try: s3_client = Session().client( "s3", aws_access_key_id=config.aws_access_key_id or None, aws_secret_access_key=config.aws_secret_access_key or None, aws_session_token=config.aws_session_token or None, ) cls.GLOBAL_S3_CLIENTS[config.aws_access_key_id] = s3_client except KeyError as exception: # Sometimes creation of client fails, and we use the global client if it exists global_client = cls.GLOBAL_S3_CLIENTS.get(config.aws_access_key_id) if global_client is None: raise ValueError("Failed to create a client for download from AWS") from exception s3_client = global_client return s3_client
def _do_download(self, request: DownloadRequest, s3_client: Any) -> bytes: """Does the download from s3""" if request.url is None: raise ValueError(f"Faulty request {request}, no URL specified.") _, _, bucket_name, url_key = request.url.split("/", 3) try: response = s3_client.get_object(Bucket=bucket_name, Key=url_key, **self.boto_params) return response["Body"].read() except NoCredentialsError as exception: raise ValueError( "The requested data is in Requester Pays AWS bucket. In order to download the data please set " "your access key either in the AWS credentials file or in the sentinelhub config.toml file using " "command line:\n" "$ sentinelhub.config --aws_access_key_id <your AWS key> --aws_secret_access_key " "<your AWS secret key>" ) from exception except s3_client.exceptions.NoSuchKey as exception: raise AwsDownloadFailedException(f"File in location {request.url} is missing") from exception except s3_client.exceptions.NoSuchBucket as exception: raise ValueError(f"Aws bucket {bucket_name} does not exist") from exception
[docs] @staticmethod def is_s3_request(request: DownloadRequest) -> bool: """Checks if data has to be downloaded from AWS s3 bucket :return: `True` if url describes location at AWS s3 bucket and `False` otherwise """ return request.url is not None and request.url.startswith("s3://")