url2uri function

one/remote/aws.py

@@ -82,6 +82,31 @@ def get_s3_virtual_host(uri, region) -> str:
     return 'https://' + '/'.join((hostname, *key))
 
 
+def url2uri(data_path, return_location=False):
+    """
+    Convert a generic Amazon virtual host URL to an S3 URI.
+
+    Parameters
+    ----------
+    data_path : str
+        An Amazon virtual host URL to convert.
+    return_location : bool
+        If true, additionally returns the location string.
+
+    Returns
+    -------
+    str
+        An S3 URI with scheme 's3://'.
+    str
+        If return_location is true, returns the bucket location, e.g. 'eu-east-1'.
+    """
+    parsed = urllib.parse.urlparse(data_path)
+    assert parsed.netloc and parsed.scheme and parsed.path
+    bucket_name, _, loc, *_ = parsed.netloc.split('.')
+    uri = f's3://{bucket_name}{parsed.path}'
+    return (uri, loc) if return_location else uri
+
+
 def is_folder(obj_summery) -> bool:
     """
     Given an S3 ObjectSummery instance, returns true if the associated object is a directory.

one/tests/remote/test_aws.py

@@ -120,7 +120,7 @@ class TestUtils(unittest.TestCase):
     """Tests for one.remote.aws utility functions"""
 
     def test_get_s3_virtual_host(self):
-        """Tests for one.remote.aws.get_s3_virtual_host function"""
+        """Tests for one.remote.aws.get_s3_virtual_host function."""
         expected = 'https://my-s3-bucket.s3.eu-east-1.amazonaws.com/'
         url = aws.get_s3_virtual_host('s3://my-s3-bucket', 'eu-east-1')
         self.assertEqual(expected, url)
@@ -135,6 +135,14 @@ def test_get_s3_virtual_host(self):
         with self.assertRaises(AssertionError):
             aws.get_s3_virtual_host('s3://my-s3-bucket/path/to/file', 'wrong-foo-4')
 
+    def test_url2uri(self):
+        """Tests for one.remote.aws.url2uri function."""
+        url = 'https://my-s3-bucket.s3.eu-east-1.amazonaws.com/path/to/file'
+        expected = 's3://my-s3-bucket/path/to/file'
+        self.assertEqual(aws.url2uri(url), expected)
+        uri, loc = aws.url2uri(url, return_location=True)
+        self.assertEqual(loc, 'eu-east-1')
+
 
 if __name__ == '__main__':
     unittest.main(exit=False)