feat: add helper for bucket bound hostname URLs

google/cloud/storage/_helpers.py

@@ -312,3 +312,28 @@ def _convert_to_timestamp(value):
     utc_naive = value.replace(tzinfo=None) - value.utcoffset()
     mtime = (utc_naive - datetime(1970, 1, 1)).total_seconds()
     return mtime
+
+
+def _bucket_bound_hostname_url(host, scheme=None, end_slash=False):
+    """Helper to build bucket bound hostname URL.
+
+    :type host: str
+    :param host: Host name.
+
+    :type scheme: str
+    :param scheme: (Optional) Web scheme. If passed, use it
+                   as a scheme in the result URL.
+
+    :type end_slash: bool
+    :param end_slash: (Optional) If True, add "/" slash to
+                      the end of the result URL.
+
+    :rtype: str
+    :returns: A bucket bound hostname URL.
+    """
+    if ":" in host:
+        return host
+
+    return "{scheme}://{host}{slash}".format(
+        scheme=scheme, host=host, slash="/" if end_slash else ""
+    )

google/cloud/storage/blob.py

@@ -56,6 +56,7 @@
 from google.cloud.exceptions import NotFound
 from google.cloud.storage._helpers import _PropertyMixin
 from google.cloud.storage._helpers import _scalar_property
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage._helpers import _convert_to_timestamp
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
@@ -514,12 +515,9 @@ def generate_signed_url(
                 bucket_name=self.bucket.name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                api_access_endpoint = bucket_bound_hostname
-            else:
-                api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             resource = "/{bucket_name}/{quoted_name}".format(
                 bucket_name=self.bucket.name, quoted_name=quoted_name

google/cloud/storage/bucket.py

@@ -37,6 +37,7 @@
 from google.cloud.storage._helpers import _validate_name
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage.acl import BucketACL
 from google.cloud.storage.acl import DefaultObjectACL
 from google.cloud.storage.blob import Blob
@@ -2560,12 +2561,9 @@ def generate_signed_url(
                 bucket_name=self.name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                api_access_endpoint = bucket_bound_hostname
-            else:
-                api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             resource = "/{bucket_name}".format(bucket_name=self.name)
 

google/cloud/storage/client.py

@@ -30,6 +30,7 @@
 from google.cloud.client import ClientWithProject
 from google.cloud.exceptions import NotFound
 from google.cloud.storage._helpers import _get_storage_host
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage._http import Connection
 from google.cloud.storage._signing import (
     get_expiration_seconds_v4,
@@ -1033,15 +1034,10 @@ def generate_signed_post_policy_v4(
         # designate URL
         if virtual_hosted_style:
             url = "https://{}.storage.googleapis.com/".format(bucket_name)
-
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:  # URL includes scheme
-                url = bucket_bound_hostname
-
-            else:  # scheme is given separately
-                url = "{scheme}://{host}/".format(
-                    scheme=scheme, host=bucket_bound_hostname
-                )
+            url = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme, end_slash=True
+            )
         else:
             url = "https://storage.googleapis.com/{}/".format(bucket_name)
 

tests/unit/test__helpers.py

@@ -343,6 +343,33 @@ def read(self, block_size):
         self.assertEqual(MD5.hash_obj._blocks, [BYTES_TO_SIGN])
 
 
+class Test__bucket_bound_hostname_url(unittest.TestCase):
+    def _call_fut(self, **args):
+        from google.cloud.storage._helpers import _bucket_bound_hostname_url
+
+        return _bucket_bound_hostname_url(**args)
+
+    def test_full_hostname(self):
+        HOST = "scheme://domain.tcl"
+        self.assertEqual(self._call_fut(host=HOST), HOST)
+
+    def test_hostname_and_scheme(self):
+        HOST = "domain.tcl"
+        SCHEME = "scheme"
+        EXPECTED_URL = SCHEME + "://" + HOST
+
+        self.assertEqual(self._call_fut(host=HOST, scheme=SCHEME), EXPECTED_URL)
+
+    def test_with_end_slash(self):
+        HOST = "domain.tcl"
+        SCHEME = "scheme"
+        EXPECTED_URL = "{scheme}://{host}/".format(scheme=SCHEME, host=HOST)
+
+        self.assertEqual(
+            self._call_fut(host=HOST, scheme=SCHEME, end_slash=True), EXPECTED_URL
+        )
+
+
 class _Connection(object):
     def __init__(self, *responses):
         self._responses = responses