diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 index 6bdc60632337..aa450ddd327f 100644 --- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 +++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 @@ -40,6 +40,7 @@ try: # we bump the minimum supported version of google-auth. from google.auth.transport.mtls import should_use_client_cert # type: ignore except ImportError: # pragma: NO COVER + {# The fallback implementation for `should_use_client_cert` can be removed once the minimum version of `google-auth` is bumped to 2.56.0 in `setup.py.j2`. #} def should_use_client_cert(): """Returns whether client certificate should be used for mTLS.""" use_client_cert = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower() @@ -51,8 +52,7 @@ except ImportError: # pragma: NO COVER return use_client_cert == "true" -{# TODO(https://github.com/googleapis/google-cloud-python/issues/17883): -Defer to google.auth.transport.mtls.should_use_mtls_endpoint (available in google-auth >= 2.56.0) +{# Defer to google.auth.transport.mtls.should_use_mtls_endpoint (available in google-auth >= 2.56.0) to parse GOOGLE_API_USE_MTLS_ENDPOINT when available in minimum supported google-auth. #} def read_environment_variables(): """Returns the environment variables used by the client. @@ -80,248 +80,272 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + {# The fallback implementation for `get_default_mtls_endpoint` can be removed once the minimum version of `google-api-core` is bumped to 2.33.0 in `setup.py.j2`. #} + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. + host = parsed.hostname + if not host: + return api_endpoint - Args: - api_endpoint (Optional[str]): the api endpoint to convert. + port = f":{parsed.port}" if parsed.port else "" - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. - - Returns: - str: The API endpoint to be used by the client. + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." - ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) - Returns: - str: The universe domain to be used by the client. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + {# The fallback implementation for `get_api_endpoint` can be removed once the minimum version of `google-api-core` is bumped to 2.33.0 in `setup.py.j2`. #} + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + {# The fallback implementation for `get_universe_domain` can be removed once the minimum version of `google-api-core` is bumped to 2.33.0 in `setup.py.j2`. #} + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - if not resolved: - raise EmptyUniverseError() - return resolved + if not resolved: + raise EmptyUniverseError() + return resolved {% if has_auto_populated_fields %} -def setup_request_id( - request: Union[google.protobuf.message.Message, "proto.Message", dict, None], - field_name: str, - is_proto3_optional: bool, -) -> None: - """Populate a UUID4 field in the request if it is not already set. - - This helper is used to ensure request idempotency by automatically - generating a unique identifier (such as `request_id`) for requests - that support it. If a request is retried, the same identifier can be - sent on subsequent retries, allowing the server to recognize the retried - request and prevent duplicate processing (e.g., creating duplicate - resources). - - Args: - request (Union[google.protobuf.message.Message, proto.Message, dict, None]): The - request object or dictionary. - field_name (str): The name of the field to populate (e.g., "request_id"). - is_proto3_optional (bool): Whether the field supports explicit presence - (defined with `optional` in proto3 syntax). When True, empty strings ("") - are preserved as explicit user input per AIP-4235, and UUID auto-population - occurs only if the field is unset. When False, any empty or falsy value is - populated with a UUID. - """ - if request is None: - return - - # Evaluate whether the field is considered "unset" and needs auto-population. - # - # According to AIP-4235, optional request ID fields must be populated - # if and only if they have explicit presence (`is_proto3_optional=True`) - # and were not set by the user (i.e. unset). Explicitly provided empty - # strings ('') must be preserved when `is_proto3_optional=True`. - should_populate = False - if isinstance(request, dict): - if is_proto3_optional: - # Case 1a: Dictionary request with explicit presence (`is_proto3_optional=True`). - # Per AIP-4235, auto-populate only if the key is completely missing from - # the dictionary or its value is explicitly set to None. - # An explicit empty string ('') must NOT be overwritten. - should_populate = field_name not in request or request[field_name] is None - else: - # Case 1b: Dictionary request without explicit presence (`is_proto3_optional=False`). - # Auto-populate if the key is missing, None, or falsy (e.g., empty string ''). - should_populate = not request.get(field_name) - else: - # Case 2: Object request (proto-plus wrapper or pure protobuf message). - if is_proto3_optional: - # Extract the protobuf from proto-plus if wrapped. - pure_pb: google.protobuf.message.Message = getattr(request, "_pb", request) - try: - should_populate = not pure_pb.HasField(field_name) - except (AttributeError, ValueError): - # Fall back if `HasField` fails or is unsupported. - should_populate = getattr(pure_pb, field_name, None) is None - else: - # Case 2b: Object request without explicit presence (`is_proto3_optional=False`). - # Auto-populate if the field value is falsy (None or empty string ''). - should_populate = not bool(getattr(request, field_name, False)) - - # If the field was found to be empty, set random id - if should_populate: - generated_id = str(uuid.uuid4()) +try: + import google.api_core + from google.api_core.gapic_v1.requests import setup_request_id + if tuple(map(int, google.api_core.__version__.split(".")[:2])) < (2, 34): # pragma: NO COVER + # google-api-core < 2.34.0 had a bug in setup_request_id for proto-plus messages. + # Fall back to the local compatibility implementation for older versions. + raise ImportError +except ImportError: # pragma: NO COVER + {# The fallback implementation for `setup_request_id` can be removed once the minimum version of `google-api-core` is bumped to 2.34.0 in `setup.py.j2`. #} + def setup_request_id( + request: Union[google.protobuf.message.Message, "proto.Message", dict, None], + field_name: str, + is_proto3_optional: bool, + ) -> None: + """Populate a UUID4 field in the request if it is not already set. + + This helper is used to ensure request idempotency by automatically + generating a unique identifier (such as `request_id`) for requests + that support it. If a request is retried, the same identifier can be + sent on subsequent retries, allowing the server to recognize the retried + request and prevent duplicate processing (e.g., creating duplicate + resources). + + Args: + request (Union[google.protobuf.message.Message, proto.Message, dict, None]): The + request object or dictionary. + field_name (str): The name of the field to populate (e.g., "request_id"). + is_proto3_optional (bool): Whether the field supports explicit presence + (defined with `optional` in proto3 syntax). When True, empty strings ("") + are preserved as explicit user input per AIP-4235, and UUID auto-population + occurs only if the field is unset. When False, any empty or falsy value is + populated with a UUID. + """ + if request is None: + return + + # Evaluate whether the field is considered "unset" and needs auto-population. + # + # According to AIP-4235, optional request ID fields must be populated + # if and only if they have explicit presence (`is_proto3_optional=True`) + # and were not set by the user (i.e. unset). Explicitly provided empty + # strings ('') must be preserved when `is_proto3_optional=True`. + should_populate = False if isinstance(request, dict): - request[field_name] = generated_id + if is_proto3_optional: + # Case 1a: Dictionary request with explicit presence (`is_proto3_optional=True`). + # Per AIP-4235, auto-populate only if the key is completely missing from + # the dictionary or its value is explicitly set to None. + # An explicit empty string ('') must NOT be overwritten. + should_populate = field_name not in request or request[field_name] is None + else: + # Case 1b: Dictionary request without explicit presence (`is_proto3_optional=False`). + # Auto-populate if the key is missing, None, or falsy (e.g., empty string ''). + should_populate = not request.get(field_name) else: - setattr(request, field_name, generated_id) + # Case 2: Object request (proto-plus wrapper or pure protobuf message). + if is_proto3_optional: + # Extract the protobuf from proto-plus if wrapped. + pure_pb: google.protobuf.message.Message = getattr(request, "_pb", request) + try: + should_populate = not pure_pb.HasField(field_name) + except (AttributeError, ValueError): + # Fall back if `HasField` fails or is unsupported. + should_populate = getattr(pure_pb, field_name, None) is None + else: + # Case 2b: Object request without explicit presence (`is_proto3_optional=False`). + # Auto-populate if the field value is falsy (None or empty string ''). + should_populate = not bool(getattr(request, field_name, False)) + + # If the field was found to be empty, set random id + if should_populate: + generated_id = str(uuid.uuid4()) + if isinstance(request, dict): + request[field_name] = generated_id + else: + setattr(request, field_name, generated_id) {% endif %} -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + {# The fallback implementation for `transcode_request` can be removed once the minimum version of `google-api-core` is bumped to 2.34.0 in `setup.py.j2`. #} + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], use_integers_for_enums=rest_numeric_enums, ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) + ) + + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json {% endblock %} diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 index f1ea148cc027..01407a160d99 100644 --- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 +++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 @@ -442,7 +442,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta): self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = {{ service.client_name }}._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe={{ service.client_name }}._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py index 1121cb71b64e..ffc75791c484 100755 --- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py +++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py @@ -491,7 +491,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = AssetServiceClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=AssetServiceClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py index 835c762fe776..da065db5907b 100755 --- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py +++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py @@ -428,7 +428,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = IAMCredentialsClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=IAMCredentialsClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py index f7027b05bcbd..f5442cba6179 100755 --- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py +++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py @@ -611,7 +611,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = EventarcClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=EventarcClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py index 62ea80691320..2ec9186dedc1 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py @@ -484,7 +484,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = ConfigServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=ConfigServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py index 2507e84cd049..dfaf6928a16d 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py @@ -415,7 +415,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = LoggingServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=LoggingServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py index 4c96b9b45d0a..7319be93a38c 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py @@ -416,7 +416,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = MetricsServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=MetricsServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py index 52cb67262f88..e136bf06d85d 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py @@ -484,7 +484,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = BaseConfigServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=BaseConfigServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py index 2507e84cd049..dfaf6928a16d 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py @@ -415,7 +415,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = LoggingServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=LoggingServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py index 6e6fe990f6a2..46949c293cd9 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py @@ -416,7 +416,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = BaseMetricsServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=BaseMetricsServiceV2Client._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py index dd14380a6c91..7b2e7759cd73 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py +++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py @@ -456,7 +456,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = CloudRedisClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=CloudRedisClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py index cd597b06c87b..a6d3f9fbb31f 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py @@ -69,175 +69,186 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) + + host = parsed.hostname + if not host: + return api_endpoint + + port = f":{parsed.port}" if parsed.port else "" + + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint + + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) + + if not has_scheme: + return urlunparse(new_parsed)[2:] + else: + return urlunparse(new_parsed) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint + else: + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - str: The API endpoint to be used by the client. + if not resolved: + raise EmptyUniverseError() + return resolved - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], + use_integers_for_enums=rest_numeric_enums, ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. - - Returns: - str: The universe domain to be used by the client. - - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") - - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - - transcoded_request = path_template.transcode(http_options, pb_request) - - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], - use_integers_for_enums=rest_numeric_enums, + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py index 1c2a3de6f32d..771b0baa9989 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py +++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py @@ -456,7 +456,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = CloudRedisClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=CloudRedisClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation. diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py index 87e049d32472..d7096741a7f9 100755 --- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py +++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py @@ -75,244 +75,263 @@ def read_environment_variables(): DEFAULT_UNIVERSE = "googleapis.com" +try: + from google.api_core.universe import get_default_mtls_endpoint +except ImportError: # pragma: NO COVER + def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: + """Converts api endpoint to mTLS endpoint. + + Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to + "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. + Other URLs (including those that do not match these domain suffixes or + already contain '.mtls.') are passed through as-is. + + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint or ".mtls." in api_endpoint.lower(): + return api_endpoint + + has_scheme = "://" in api_endpoint + if not has_scheme: + parsed = urlparse("//" + api_endpoint) + else: + parsed = urlparse(api_endpoint) -def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: - """Converts api endpoint to mTLS endpoint. - - Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to - "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. - Other URLs (including those that do not match these domain suffixes or - already contain '.mtls.') are passed through as-is. - - Args: - api_endpoint (Optional[str]): the api endpoint to convert. - - Returns: - Optional[str]: converted mTLS api endpoint. - """ - if not api_endpoint or ".mtls." in api_endpoint.lower(): - return api_endpoint - - has_scheme = "://" in api_endpoint - if not has_scheme: - parsed = urlparse("//" + api_endpoint) - else: - parsed = urlparse(api_endpoint) - - host = parsed.hostname - if not host: - return api_endpoint - - port = f":{parsed.port}" if parsed.port else "" - - lowered_host = host.lower() - suffix_sandbox = ".sandbox.googleapis.com" - suffix_google = ".googleapis.com" - if lowered_host.endswith(suffix_sandbox): - new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" - elif lowered_host.endswith(suffix_google): - new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" - else: - return api_endpoint - - netloc = new_host + port - new_parsed = parsed._replace(netloc=netloc) - - if not has_scheme: - return urlunparse(new_parsed)[2:] - else: - return urlunparse(new_parsed) - -def get_api_endpoint( - api_override: Optional[str], - universe_domain: str, - default_universe: str, - default_mtls_endpoint: Optional[str], - default_endpoint_template: str, - use_mtls: bool, -) -> str: - """Return the API endpoint used by the client. - - Args: - api_override (Optional[str]): The API endpoint override. If specified, - this is always returned. - universe_domain (str): The universe domain used by the client. - default_universe (str): The default universe domain. - default_mtls_endpoint (Optional[str]): The default mTLS endpoint. - default_endpoint_template (str): The default endpoint template containing - a placeholder `{UNIVERSE_DOMAIN}`. - use_mtls (bool): Whether to use the mTLS endpoint. - - Returns: - str: The API endpoint to be used by the client. + host = parsed.hostname + if not host: + return api_endpoint - Raises: - google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but - not supported in the configured universe domain. - ValueError: If mTLS is requested but no mTLS endpoint is available. - """ - if api_override is not None: - return api_override + port = f":{parsed.port}" if parsed.port else "" - if use_mtls: - if universe_domain.lower() != default_universe.lower(): - raise MutualTLSChannelError( - f"mTLS is not supported in any universe other than {default_universe}." - ) - if not default_mtls_endpoint: - raise ValueError("mTLS endpoint is not available.") - return default_mtls_endpoint - else: - return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) - -def get_universe_domain( - *potential_universes: Optional[str], - default_universe: str = DEFAULT_UNIVERSE, -) -> str: - """Return the universe domain used by the client. - - Args: - *potential_universes (Optional[str]): Potential universe domains in order of preference. - default_universe (str): The default universe domain. + lowered_host = host.lower() + suffix_sandbox = ".sandbox.googleapis.com" + suffix_google = ".googleapis.com" + if lowered_host.endswith(suffix_sandbox): + new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" + elif lowered_host.endswith(suffix_google): + new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" + else: + return api_endpoint - Returns: - str: The universe domain to be used by the client. + netloc = new_host + port + new_parsed = parsed._replace(netloc=netloc) - Raises: - EmptyUniverseError: If the resolved universe domain is an empty string. - """ - resolved = next( - (x.strip() for x in potential_universes if x is not None), - default_universe, - ) - - if not resolved: - raise EmptyUniverseError() - return resolved - - -def setup_request_id( - request: Union[google.protobuf.message.Message, "proto.Message", dict, None], - field_name: str, - is_proto3_optional: bool, -) -> None: - """Populate a UUID4 field in the request if it is not already set. - - This helper is used to ensure request idempotency by automatically - generating a unique identifier (such as `request_id`) for requests - that support it. If a request is retried, the same identifier can be - sent on subsequent retries, allowing the server to recognize the retried - request and prevent duplicate processing (e.g., creating duplicate - resources). - - Args: - request (Union[google.protobuf.message.Message, proto.Message, dict, None]): The - request object or dictionary. - field_name (str): The name of the field to populate (e.g., "request_id"). - is_proto3_optional (bool): Whether the field supports explicit presence - (defined with `optional` in proto3 syntax). When True, empty strings ("") - are preserved as explicit user input per AIP-4235, and UUID auto-population - occurs only if the field is unset. When False, any empty or falsy value is - populated with a UUID. - """ - if request is None: - return - - # Evaluate whether the field is considered "unset" and needs auto-population. - # - # According to AIP-4235, optional request ID fields must be populated - # if and only if they have explicit presence (`is_proto3_optional=True`) - # and were not set by the user (i.e. unset). Explicitly provided empty - # strings ('') must be preserved when `is_proto3_optional=True`. - should_populate = False - if isinstance(request, dict): - if is_proto3_optional: - # Case 1a: Dictionary request with explicit presence (`is_proto3_optional=True`). - # Per AIP-4235, auto-populate only if the key is completely missing from - # the dictionary or its value is explicitly set to None. - # An explicit empty string ('') must NOT be overwritten. - should_populate = field_name not in request or request[field_name] is None + if not has_scheme: + return urlunparse(new_parsed)[2:] else: - # Case 1b: Dictionary request without explicit presence (`is_proto3_optional=False`). - # Auto-populate if the key is missing, None, or falsy (e.g., empty string ''). - should_populate = not request.get(field_name) - else: - # Case 2: Object request (proto-plus wrapper or pure protobuf message). - if is_proto3_optional: - # Extract the protobuf from proto-plus if wrapped. - pure_pb: google.protobuf.message.Message = getattr(request, "_pb", request) - try: - should_populate = not pure_pb.HasField(field_name) - except (AttributeError, ValueError): - # Fall back if `HasField` fails or is unsupported. - should_populate = getattr(pure_pb, field_name, None) is None - else: - # Case 2b: Object request without explicit presence (`is_proto3_optional=False`). - # Auto-populate if the field value is falsy (None or empty string ''). - should_populate = not bool(getattr(request, field_name, False)) + return urlunparse(new_parsed) - # If the field was found to be empty, set random id - if should_populate: - generated_id = str(uuid.uuid4()) - if isinstance(request, dict): - request[field_name] = generated_id +try: + from google.api_core.universe import get_api_endpoint +except ImportError: # pragma: NO COVER + def get_api_endpoint( + api_override: Optional[str], + universe_domain: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, + use_mtls: bool, + ) -> str: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. If specified, + this is always returned. + universe_domain (str): The universe domain used by the client. + default_universe (str): The default universe domain. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template containing + a placeholder `{UNIVERSE_DOMAIN}`. + use_mtls (bool): Whether to use the mTLS endpoint. + + Returns: + str: The API endpoint to be used by the client. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but + not supported in the configured universe domain. + ValueError: If mTLS is requested but no mTLS endpoint is available. + """ + if api_override is not None: + return api_override + + if use_mtls: + if universe_domain.lower() != default_universe.lower(): + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than {default_universe}." + ) + if not default_mtls_endpoint: + raise ValueError("mTLS endpoint is not available.") + return default_mtls_endpoint else: - setattr(request, field_name, generated_id) - + return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) -def transcode_request( - http_options: List[Dict[str, str]], - request: Any, - required_fields_default_values: Optional[Dict[str, Any]] = None, - rest_numeric_enums: bool = False, -) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: - """Transcodes a request into HTTP method, URI, body, and query parameters. - - Args: - http_options (List[Dict[str, str]]): List of HTTP transcoding rules. - request (Any): The protobuf or proto-plus request message. - required_fields_default_values (Optional[Dict[str, Any]]): Dictionary - of required fields default values to merge into query parameters if missing. - rest_numeric_enums (bool): Whether to encode enums as integers. +try: + from google.api_core.universe import get_universe_domain +except ImportError: # pragma: NO COVER + def get_universe_domain( + *potential_universes: Optional[str], + default_universe: str, + ) -> str: + """Return the universe domain used by the client. + + Args: + *potential_universes (Optional[str]): Potential universe domains in order of preference. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + EmptyUniverseError: If the resolved universe domain is an empty string. + """ + resolved = next( + (x.strip() for x in potential_universes if x is not None), + default_universe, + ) - Returns: - Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: - - The raw transcoded request dictionary (containing keys like 'uri', 'method'). - - The serialized request body JSON string, or None if no body. - - The query parameters dictionary. - """ - if request is None: - raise TypeError("request cannot be None") + if not resolved: + raise EmptyUniverseError() + return resolved - # Convert proto-plus message to its underlying protobuf message if needed - pb_request = getattr(request, "_pb", request) - transcoded_request = path_template.transcode(http_options, pb_request) +try: + import google.api_core + from google.api_core.gapic_v1.requests import setup_request_id + if tuple(map(int, google.api_core.__version__.split(".")[:2])) < (2, 34): # pragma: NO COVER + # google-api-core < 2.34.0 had a bug in setup_request_id for proto-plus messages. + # Fall back to the local compatibility implementation for older versions. + raise ImportError +except ImportError: # pragma: NO COVER + def setup_request_id( + request: Union[google.protobuf.message.Message, "proto.Message", dict, None], + field_name: str, + is_proto3_optional: bool, + ) -> None: + """Populate a UUID4 field in the request if it is not already set. + + This helper is used to ensure request idempotency by automatically + generating a unique identifier (such as `request_id`) for requests + that support it. If a request is retried, the same identifier can be + sent on subsequent retries, allowing the server to recognize the retried + request and prevent duplicate processing (e.g., creating duplicate + resources). + + Args: + request (Union[google.protobuf.message.Message, proto.Message, dict, None]): The + request object or dictionary. + field_name (str): The name of the field to populate (e.g., "request_id"). + is_proto3_optional (bool): Whether the field supports explicit presence + (defined with `optional` in proto3 syntax). When True, empty strings ("") + are preserved as explicit user input per AIP-4235, and UUID auto-population + occurs only if the field is unset. When False, any empty or falsy value is + populated with a UUID. + """ + if request is None: + return + + # Evaluate whether the field is considered "unset" and needs auto-population. + # + # According to AIP-4235, optional request ID fields must be populated + # if and only if they have explicit presence (`is_proto3_optional=True`) + # and were not set by the user (i.e. unset). Explicitly provided empty + # strings ('') must be preserved when `is_proto3_optional=True`. + should_populate = False + if isinstance(request, dict): + if is_proto3_optional: + # Case 1a: Dictionary request with explicit presence (`is_proto3_optional=True`). + # Per AIP-4235, auto-populate only if the key is completely missing from + # the dictionary or its value is explicitly set to None. + # An explicit empty string ('') must NOT be overwritten. + should_populate = field_name not in request or request[field_name] is None + else: + # Case 1b: Dictionary request without explicit presence (`is_proto3_optional=False`). + # Auto-populate if the key is missing, None, or falsy (e.g., empty string ''). + should_populate = not request.get(field_name) + else: + # Case 2: Object request (proto-plus wrapper or pure protobuf message). + if is_proto3_optional: + # Extract the protobuf from proto-plus if wrapped. + pure_pb: google.protobuf.message.Message = getattr(request, "_pb", request) + try: + should_populate = not pure_pb.HasField(field_name) + except (AttributeError, ValueError): + # Fall back if `HasField` fails or is unsupported. + should_populate = getattr(pure_pb, field_name, None) is None + else: + # Case 2b: Object request without explicit presence (`is_proto3_optional=False`). + # Auto-populate if the field value is falsy (None or empty string ''). + should_populate = not bool(getattr(request, field_name, False)) + + # If the field was found to be empty, set random id + if should_populate: + generated_id = str(uuid.uuid4()) + if isinstance(request, dict): + request[field_name] = generated_id + else: + setattr(request, field_name, generated_id) - body_json = None - if transcoded_request.get("body") is not None: - body_json = json_format.MessageToJson( - transcoded_request["body"], - use_integers_for_enums=rest_numeric_enums, - ) - query_params_json = {} - if transcoded_request.get("query_params") is not None: - query_params_json = json.loads( - json_format.MessageToJson( - transcoded_request["query_params"], +try: + from google.api_core.rest_helpers import transcode_request # type: ignore +except ImportError: # pragma: NO COVER + def transcode_request( + http_options: List[Dict[str, str]], + request: Any, + required_fields_default_values: Optional[Dict[str, Any]] = None, + rest_numeric_enums: bool = False, + ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: + """Transcodes a request into HTTP method, URI, body, and query parameters. + + Args: + http_options (List[Dict[str, str]]): List of HTTP transcoding rules. + request (Any): The protobuf or proto-plus request message. + required_fields_default_values (Optional[Dict[str, Any]]): Dictionary + of required fields default values to merge into query parameters if missing. + rest_numeric_enums (bool): Whether to encode enums as integers. + + Returns: + Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: + - The raw transcoded request dictionary (containing keys like 'uri', 'method'). + - The serialized request body JSON string, or None if no body. + - The query parameters dictionary. + """ + if request is None: + raise TypeError("request cannot be None") + + # Convert proto-plus message to its underlying protobuf message if needed + pb_request = getattr(request, "_pb", request) + + transcoded_request = path_template.transcode(http_options, pb_request) + + body_json = None + if transcoded_request.get("body") is not None: + body_json = json_format.MessageToJson( + transcoded_request["body"], use_integers_for_enums=rest_numeric_enums, ) - ) - # If required_fields_default_values is provided, we merge default values for missing - # required fields into the query parameters. - if required_fields_default_values: - for k, v in required_fields_default_values.items(): - if k not in query_params_json: - query_params_json[k] = v + query_params_json = {} + if transcoded_request.get("query_params") is not None: + query_params_json = json.loads( + json_format.MessageToJson( + transcoded_request["query_params"], + use_integers_for_enums=rest_numeric_enums, + ) + ) + + # If required_fields_default_values is provided, we merge default values for missing + # required fields into the query parameters. + if required_fields_default_values: + for k, v in required_fields_default_values.items(): + if k not in query_params_json: + query_params_json[k] = v - if rest_numeric_enums: - query_params_json["$alt"] = "json;enum-encoding=int" + if rest_numeric_enums: + query_params_json["$alt"] = "json;enum-encoding=int" - return transcoded_request, body_json, query_params_json + return transcoded_request, body_json, query_params_json diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py index c066f52b0b1c..ee8cac5e7107 100755 --- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py +++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py @@ -452,7 +452,7 @@ def __init__(self, *, self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables() self._client_cert_source = StorageBatchOperationsClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert) - self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env) + self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=StorageBatchOperationsClient._DEFAULT_UNIVERSE) self._api_endpoint: str = "" # updated below, depending on `transport` # Initialize the universe domain validation.