diff --git a/one/remote/aws.py b/one/remote/aws.py index d08654fa..0c9b41dc 100644 --- a/one/remote/aws.py +++ b/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. diff --git a/one/tests/remote/test_aws.py b/one/tests/remote/test_aws.py index 542226e5..fc820c89 100644 --- a/one/tests/remote/test_aws.py +++ b/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)