Storage: Add Blob.from_string and Bucket.from_string.

storage/google/cloud/storage/blob.py

@@ -310,6 +310,37 @@ def public_url(self):
             quoted_name=_quote(self.name, safe=b"/~"),
         )
 
+    @classmethod
+    def from_string(cls, uri, client=None):
+        """Get a constructor for blob object by URI.
+
+         :type uri: str
+         :param uri: The blob uri pass to get blob object.
+
+         :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+         :param client: Optional. The client to use.
+
+         :rtype: :class:`google.cloud.storage.blob.Blob`
+         :returns: The blob object created.
+
+         Example:
+            Get a constructor for blob object by URI..
+
+            >>> from google.cloud import storage
+            >>> from google.cloud.storage.blob import Blob
+            >>> client = storage.Client()
+            >>> blob = Blob.from_string("gs://bucket/object")
+         """
+        from google.cloud.storage.bucket import Bucket
+
+        scheme, netloc, path, query, frag = urlsplit(uri)
+        if scheme != "gs":
+            raise ValueError("URI scheme must be gs")
+
+        bucket = Bucket(client, name=netloc)
+        return cls(path[1:], bucket)
+
     def generate_signed_url(
         self,
         expiration=None,

storage/google/cloud/storage/bucket.py

@@ -21,6 +21,7 @@
 import warnings
 
 import six
+from six.moves.urllib.parse import urlsplit
 
 from google.api_core import page_iterator
 from google.api_core import datetime_helpers
@@ -495,6 +496,35 @@ def user_project(self):
         """
         return self._user_project
 
+    @classmethod
+    def from_string(cls, uri, client=None):
+        """Get a constructor for bucket object by URI.
+
+        :type uri: str
+        :param uri: The bucket uri pass to get bucket object.
+
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: Optional. The client to use.
+
+        :rtype: :class:`google.cloud.storage.bucket.Bucket`
+        :returns: The bucket object created.
+
+        Example:
+            Get a constructor for bucket object by URI..
+
+            >>> from google.cloud import storage
+            >>> from google.cloud.storage.bucket import Bucket
+            >>> client = storage.Client()
+            >>> bucket = Bucket.from_string("gs://bucket",client)
+        """
+        scheme, netloc, path, query, frag = urlsplit(uri)
+
+        if scheme != "gs":
+            raise ValueError("URI scheme must be gs")
+
+        return cls(client, name=netloc)
+
     def blob(
         self,
         blob_name,

storage/google/cloud/storage/client.py

@@ -16,8 +16,6 @@
 
 import google.api_core.client_options
 
-from six.moves.urllib.parse import urlsplit
-
 from google.auth.credentials import AnonymousCredentials
 
 from google.api_core import page_iterator
@@ -420,13 +418,8 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
         try:
             blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
         except AttributeError:
-            scheme, netloc, path, query, frag = urlsplit(blob_or_uri)
-            if scheme != "gs":
-                raise ValueError("URI scheme must be gs")
-            bucket = Bucket(self, name=netloc)
-            blob_or_uri = Blob(path[1:], bucket)
-
-            blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
+            blob = Blob.from_string(blob_or_uri)
+            blob.download_to_file(file_obj, client=self, start=start, end=end)
 
     def list_blobs(
         self,

storage/tests/unit/test_blob.py

@@ -23,6 +23,7 @@
 
 import google.cloud.storage.blob
 import mock
+import pytest
 import six
 from six.moves import http_client
 
@@ -3171,6 +3172,41 @@ def test_updated_unset(self):
         blob = self._make_one("blob-name", bucket=BUCKET)
         self.assertIsNone(blob.updated)
 
+    def test_from_string_w_valid_uri(self):
+        from google.cloud.storage.blob import Blob
+
+        connection = _Connection()
+        client = _Client(connection)
+        uri = "gs://BUCKET_NAME/b"
+        blob = Blob.from_string(uri, client)
+
+        self.assertIsInstance(blob, Blob)
+        self.assertIs(blob.client, client)
+        self.assertEqual(blob.name, "b")
+        self.assertEqual(blob.bucket.name, "BUCKET_NAME")
+
+    def test_from_string_w_invalid_uri(self):
+        from google.cloud.storage.blob import Blob
+
+        connection = _Connection()
+        client = _Client(connection)
+
+        with pytest.raises(ValueError, match="URI scheme must be gs"):
+            Blob.from_string("http://bucket_name/b", client)
+
+    def test_from_string_w_domain_name_bucket(self):
+        from google.cloud.storage.blob import Blob
+
+        connection = _Connection()
+        client = _Client(connection)
+        uri = "gs://buckets.example.com/b"
+        blob = Blob.from_string(uri, client)
+
+        self.assertIsInstance(blob, Blob)
+        self.assertIs(blob.client, client)
+        self.assertEqual(blob.name, "b")
+        self.assertEqual(blob.bucket.name, "buckets.example.com")
+
 
 class Test__quote(unittest.TestCase):
     @staticmethod

storage/tests/unit/test_bucket.py

@@ -16,6 +16,7 @@
 import unittest
 
 import mock
+import pytest
 
 
 def _create_signing_credentials():
@@ -2668,6 +2669,39 @@ def _generate_signed_url_helper(
         }
         signer.assert_called_once_with(expected_creds, **expected_kwargs)
 
+    def test_get_bucket_from_string_w_valid_uri(self):
+        from google.cloud.storage.bucket import Bucket
+
+        connection = _Connection()
+        client = _Client(connection)
+        BUCKET_NAME = "BUCKET_NAME"
+        uri = "gs://" + BUCKET_NAME
+        bucket = Bucket.from_string(uri, client)
+        self.assertIsInstance(bucket, Bucket)
+        self.assertIs(bucket.client, client)
+        self.assertEqual(bucket.name, BUCKET_NAME)
+
+    def test_get_bucket_from_string_w_invalid_uri(self):
+        from google.cloud.storage.bucket import Bucket
+
+        connection = _Connection()
+        client = _Client(connection)
+
+        with pytest.raises(ValueError, match="URI scheme must be gs"):
+            Bucket.from_string("http://bucket_name", client)
+
+    def test_get_bucket_from_string_w_domain_name_bucket(self):
+        from google.cloud.storage.bucket import Bucket
+
+        connection = _Connection()
+        client = _Client(connection)
+        BUCKET_NAME = "buckets.example.com"
+        uri = "gs://" + BUCKET_NAME
+        bucket = Bucket.from_string(uri, client)
+        self.assertIsInstance(bucket, Bucket)
+        self.assertIs(bucket.client, client)
+        self.assertEqual(bucket.name, BUCKET_NAME)
+
     def test_generate_signed_url_no_version_passed_warning(self):
         self._generate_signed_url_helper()
 

storage/tests/unit/test_client.py

@@ -676,7 +676,9 @@ def test_download_blob_to_file_with_uri(self):
         blob = mock.Mock()
         file_obj = io.BytesIO()
 
-        with mock.patch("google.cloud.storage.client.Blob", return_value=blob):
+        with mock.patch(
+            "google.cloud.storage.client.Blob.from_string", return_value=blob
+        ):
             client.download_blob_to_file("gs://bucket_name/path/to/object", file_obj)
 
         blob.download_to_file.assert_called_once_with(
@@ -687,14 +689,10 @@ def test_download_blob_to_file_with_invalid_uri(self):
         project = "PROJECT"
         credentials = _make_credentials()
         client = self._make_one(project=project, credentials=credentials)
-        blob = mock.Mock()
         file_obj = io.BytesIO()
 
-        with mock.patch("google.cloud.storage.client.Blob", return_value=blob):
-            with pytest.raises(ValueError, match="URI scheme must be gs"):
-                client.download_blob_to_file(
-                    "http://bucket_name/path/to/object", file_obj
-                )
+        with pytest.raises(ValueError, match="URI scheme must be gs"):
+            client.download_blob_to_file("http://bucket_name/path/to/object", file_obj)
 
     def test_list_blobs(self):
         from google.cloud.storage.bucket import Bucket