Allow multiple S3 endpoints

Allow S3 URLs in the form "s3://profile@bucket/...", with profiles configured via environment variables LSST_RESOURCES_S3_PROFILE_<profile>.  This allows users to access multiple S3 services simultaneously.

Author: dhirving

python/lsst/resources/s3.py

@@ -192,13 +192,30 @@ def _transfer_config(self) -> TransferConfig:
     @property
     def client(self) -> boto3.client:
         """Client object to address remote resource."""
-        # Defer import for circular dependencies
-        return getS3Client()
+        return getS3Client(self.profile)
+
+    @property
+    def profile(self) -> str | None:
+        return self._uri.username
+
+    @property
+    def bucket(self) -> str:
+        bucket = self._uri.hostname
+        if not bucket:
+            raise ValueError(f"S3 URI does not include bucket name: '{str(self)}'")
+
+        return bucket
 
     @classmethod
     def _mexists(cls, uris: Iterable[ResourcePath]) -> dict[ResourcePath, bool]:
         # Force client to be created before creating threads.
-        getS3Client()
+        profiles = set[str | None]()
+        for path in uris:
+            if path.scheme == "s3":
+                path = cast(S3ResourcePath, path)
+                profiles.add(path.profile)
+        for profile in profiles:
+            getS3Client(profile)
 
         return super()._mexists(uris)
 
@@ -207,16 +224,16 @@ def exists(self) -> bool:
         """Check that the S3 resource exists."""
         if self.is_root:
             # Only check for the bucket since the path is irrelevant
-            return bucketExists(self.netloc)
-        exists, _ = s3CheckFileExists(self, client=self.client)
+            return bucketExists(self.bucket, self.client)
+        exists, _ = s3CheckFileExists(self, bucket=self.bucket, client=self.client)
         return exists
 
     @backoff.on_exception(backoff.expo, retryable_io_errors, max_time=max_retry_time)
     def size(self) -> int:
         """Return the size of the resource in bytes."""
         if self.dirLike:
             return 0
-        exists, sz = s3CheckFileExists(self, client=self.client)
+        exists, sz = s3CheckFileExists(self, bucket=self.bucket, client=self.client)
         if not exists:
             raise FileNotFoundError(f"Resource {self} does not exist")
         return sz
@@ -229,7 +246,7 @@ def remove(self) -> None:
         # for checking all the keys again, reponse is  HTTP 204 OK
         # response all the time
         try:
-            self.client.delete_object(Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.delete_object(Bucket=self.bucket, Key=self.relativeToPathRoot)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError("No such resource: {self}") from err
 
@@ -239,7 +256,7 @@ def read(self, size: int = -1) -> bytes:
         if size > 0:
             args["Range"] = f"bytes=0-{size-1}"
         try:
-            response = self.client.get_object(Bucket=self.netloc, Key=self.relativeToPathRoot, **args)
+            response = self.client.get_object(Bucket=self.bucket, Key=self.relativeToPathRoot, **args)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError(f"No such resource: {self}") from err
         except ClientError as err:
@@ -255,20 +272,20 @@ def write(self, data: bytes, overwrite: bool = True) -> None:
         if not overwrite and self.exists():
             raise FileExistsError(f"Remote resource {self} exists and overwrite has been disabled")
         with time_this(log, msg="Write to %s", args=(self,)):
-            self.client.put_object(Bucket=self.netloc, Key=self.relativeToPathRoot, Body=data)
+            self.client.put_object(Bucket=self.bucket, Key=self.relativeToPathRoot, Body=data)
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
     def mkdir(self) -> None:
         """Write a directory key to S3."""
-        if not bucketExists(self.netloc):
-            raise ValueError(f"Bucket {self.netloc} does not exist for {self}!")
+        if not bucketExists(self.bucket, self.client):
+            raise ValueError(f"Bucket {self.bucket} does not exist for {self}!")
 
         if not self.dirLike:
             raise NotADirectoryError(f"Can not create a 'directory' for file-like URI {self}")
 
         # don't create S3 key when root is at the top-level of an Bucket
         if self.path != "/":
-            self.client.put_object(Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.put_object(Bucket=self.bucket, Key=self.relativeToPathRoot)
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
     def _download_file(self, local_file: IO, progress: ProgressPercentage | None) -> None:
@@ -279,7 +296,7 @@ def _download_file(self, local_file: IO, progress: ProgressPercentage | None) ->
         """
         try:
             self.client.download_fileobj(
-                self.netloc,
+                self.bucket,
                 self.relativeToPathRoot,
                 local_file,
                 Callback=progress,
@@ -324,7 +341,7 @@ def _upload_file(self, local_file: ResourcePath, progress: ProgressPercentage |
         """
         try:
             self.client.upload_file(
-                local_file.ospath, self.netloc, self.relativeToPathRoot, Callback=progress
+                local_file.ospath, self.bucket, self.relativeToPathRoot, Callback=progress
             )
         except self.client.exceptions.NoSuchBucket as err:
             raise NotADirectoryError(f"Target does not exist: {err}") from err
@@ -333,13 +350,13 @@ def _upload_file(self, local_file: ResourcePath, progress: ProgressPercentage |
             raise
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
-    def _copy_from(self, src: ResourcePath) -> None:
+    def _copy_from(self, src: S3ResourcePath) -> None:
         copy_source = {
-            "Bucket": src.netloc,
+            "Bucket": src.bucket,
             "Key": src.relativeToPathRoot,
         }
         try:
-            self.client.copy_object(CopySource=copy_source, Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.copy_object(CopySource=copy_source, Bucket=self.bucket, Key=self.relativeToPathRoot)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError("No such resource to transfer: {self}") from err
         except ClientError as err:
@@ -469,7 +486,7 @@ def walk(
         filenames = []
         files_there = False
 
-        for page in s3_paginator.paginate(Bucket=self.netloc, Prefix=prefix, Delimiter="/"):
+        for page in s3_paginator.paginate(Bucket=self.bucket, Prefix=prefix, Delimiter="/"):
             # All results are returned as full key names and we must
             # convert them back to the root form. The prefix is fixed
             # and delimited so that is a simple trim
@@ -507,7 +524,7 @@ def _openImpl(
         *,
         encoding: str | None = None,
     ) -> Iterator[ResourceHandleProtocol]:
-        with S3ResourceHandle(mode, log, self.client, self.netloc, self.relativeToPathRoot) as handle:
+        with S3ResourceHandle(mode, log, self.client, self.bucket, self.relativeToPathRoot) as handle:
             if "b" in mode:
                 yield handle
             else:
@@ -529,6 +546,6 @@ def generate_presigned_put_url(self, *, expiration_time_seconds: int) -> str:
     def _generate_presigned_url(self, method: str, expiration_time_seconds: int) -> str:
         return self.client.generate_presigned_url(
             method,
-            Params={"Bucket": self.netloc, "Key": self.relativeToPathRoot},
+            Params={"Bucket": self.bucket, "Key": self.relativeToPathRoot},
             ExpiresIn=expiration_time_seconds,
         )

python/lsst/resources/s3utils.py

@@ -34,13 +34,14 @@
 from contextlib import contextmanager
 from http.client import HTTPException, ImproperConnectionState
 from types import ModuleType
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, cast
 from unittest.mock import patch
 
 from botocore.exceptions import ClientError
 from botocore.handlers import validate_bucket_name
 from deprecated.sphinx import deprecated
 from urllib3.exceptions import HTTPError, RequestError
+from urllib3.util import Url, parse_url
 
 if TYPE_CHECKING:
     from unittest import TestCase
@@ -178,18 +179,32 @@ def clean_test_environment_for_s3() -> Iterator[None]:
         yield
 
 
-def getS3Client() -> boto3.client:
+def getS3Client(profile: str | None = None) -> boto3.client:
     """Create a S3 client with AWS (default) or the specified endpoint.
 
+    Parameters
+    ----------
+    profile : `str`, optional
+        The name of an S3 profile describing which S3 service to use.
+
     Returns
     -------
     s3client : `botocore.client.S3`
         A client of the S3 service.
 
     Notes
     -----
-    The endpoint URL is from the environment variable S3_ENDPOINT_URL.
-    If none is specified, the default AWS one is used.
+    If an explicit profile name is specified, its configuration is read from an
+    environment variable named ``LSST_RESOURCES_S3_PROFILE_<profile>``.  Note
+    that the name of the profile is case sensitive.  This configuration is
+    specified in the format:
+    ``https://<access key ID>:<secret key>@<s3 endpoint hostname>``
+
+    The access key ID and secret key are optional -- if not specified, they
+    will be looked up via the AWS credentials file.
+
+    If profile is `None`, this configuration is from the environment variable
+    S3_ENDPOINT_URL.  If none is specified, the default AWS one is used.
 
     If the environment variable LSST_DISABLE_BUCKET_VALIDATION exists
     and has a value that is not empty, "0", "f", "n", or "false"
@@ -202,26 +217,72 @@ def getS3Client() -> boto3.client:
     if botocore is None:
         raise ModuleNotFoundError("Could not find botocore. Are you sure it is installed?")
 
-    endpoint = os.environ.get("S3_ENDPOINT_URL", None)
-    if not endpoint:
-        endpoint = None  # Handle ""
+    if profile is None:
+        endpoint = os.environ.get("S3_ENDPOINT_URL", None)
+        if not endpoint:
+            endpoint = None  # Handle ""
+    else:
+        var_name = f"LSST_RESOURCES_S3_PROFILE_{profile}"
+        endpoint = os.environ.get(var_name, None)
+        if not endpoint:
+            raise RuntimeError(
+                f"No configuration found for requested S3 profile '{profile}'."
+                f" Set the environment variable '{var_name}' to configure it."
+            )
+
     disable_value = os.environ.get("LSST_DISABLE_BUCKET_VALIDATION", "0")
     skip_validation = not re.search(r"^(0|f|n|false)?$", disable_value, re.I)
 
     return _get_s3_client(endpoint, skip_validation)
 
 
 @functools.lru_cache
-def _get_s3_client(endpoint: str, skip_validation: bool) -> boto3.client:
+def _get_s3_client(endpoint: str | None, skip_validation: bool) -> boto3.client:
     # Helper function to cache the client for this endpoint
     config = botocore.config.Config(read_timeout=180, retries={"mode": "adaptive", "max_attempts": 10})
 
-    client = boto3.client("s3", endpoint_url=endpoint, config=config)
+    endpoint_config = _parse_endpoint_config(endpoint)
+
+    client = boto3.client(
+        "s3",
+        endpoint_url=endpoint_config.endpoint_url,
+        aws_access_key_id=endpoint_config.access_key_id,
+        aws_secret_access_key=endpoint_config.secret_access_key,
+        config=config,
+    )
     if skip_validation:
         client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
     return client
 
 
+class _EndpointConfig(NamedTuple):
+    endpoint_url: str | None = None
+    access_key_id: str | None = None
+    secret_access_key: str | None = None
+
+
+def _parse_endpoint_config(endpoint: str | None) -> _EndpointConfig:
+    if not endpoint:
+        return _EndpointConfig()
+
+    parsed = parse_url(endpoint)
+
+    # Strip the username/password portion of the URL from the result.
+    endpoint_url = Url(host=parsed.host, path=parsed.path, port=parsed.port, scheme=parsed.scheme).url
+
+    access_key_id = None
+    secret_access_key = None
+    if parsed.auth:
+        split = parsed.auth.split(":")
+        if len(split) != 2:
+            raise ValueError("S3 access key and secret not in expected format.")
+        access_key_id, secret_access_key = split
+
+    return _EndpointConfig(
+        endpoint_url=endpoint_url, access_key_id=access_key_id, secret_access_key=secret_access_key
+    )
+
+
 def s3CheckFileExists(
     path: Location | ResourcePath | str,
     bucket: str | None = None,
@@ -268,7 +329,8 @@ def s3CheckFileExists(
             bucket = uri.netloc
             filepath = uri.relativeToPathRoot
     elif isinstance(path, ResourcePath | Location):
-        bucket = path.netloc
+        if bucket is None:
+            bucket = path.netloc
         filepath = path.relativeToPathRoot
     else:
         raise TypeError(f"Unsupported path type: {path!r}.")