fix(generator): use unified _compat.py.j2

Author: hebaalazzeh

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -2,7 +2,139 @@
 
 """A compatibility module for older versions of google-api-core."""
 
+import os
+import re
 import uuid
+from typing import Optional, Callable, Tuple, Union
+from google.auth.exceptions import MutualTLSChannelError
+
+try:
+    from google.api_core.universe import (
+        get_default_mtls_endpoint,
+        get_api_endpoint,
+        get_universe_domain,
+    )
+except ImportError:
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint."""
+        if not api_endpoint:
+            return api_endpoint
+
+        mtls_endpoint_re = re.compile(
+            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
+        )
+
+        m = mtls_endpoint_re.match(api_endpoint)
+        if m is None:
+            # Could not parse api_endpoint; return as-is.
+            return api_endpoint
+
+        name, mtls, sandbox, googledomain = m.groups()
+        if mtls or not googledomain:
+            return api_endpoint
+
+        if sandbox:
+            return api_endpoint.replace(
+                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
+            )
+
+        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
+
+    def get_api_endpoint(
+        api_override: Optional[str],
+        client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+        universe_domain: str,
+        use_mtls_endpoint: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+    ) -> Optional[str]:
+        """Return the API endpoint used by the client."""
+        if api_override is not None:
+            api_endpoint = api_override
+        elif use_mtls_endpoint == "always" or (use_mtls_endpoint == "auto" and client_cert_source):
+            if universe_domain != default_universe:
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            api_endpoint = default_mtls_endpoint
+        else:
+            api_endpoint = default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+        return api_endpoint
+
+    def get_universe_domain(
+        client_universe_domain: Optional[str],
+        universe_domain_env: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        if client_universe_domain is not None:
+            universe_domain = client_universe_domain
+        elif universe_domain_env is not None:
+            universe_domain = universe_domain_env
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain
+
+
+try:
+    from google.api_core.gapic_v1.config import (
+        use_client_cert_effective,
+        get_client_cert_source,
+        read_environment_variables,
+    )
+except ImportError:
+    from google.auth.transport import mtls  # type: ignore
+
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def use_client_cert_effective() -> bool:
+        """Returns whether client certificate should be used for mTLS."""
+        if hasattr(mtls, "should_use_client_cert"):
+            return mtls.should_use_client_cert()
+        else:
+            use_client_cert_str = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
+            if use_client_cert_str not in ("true", "false"):
+                raise ValueError(
+                    "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
+                    " either `true` or `false`"
+                )
+            return use_client_cert_str == "true"
+
+    def get_client_cert_source(
+        provided_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+        use_cert_flag: bool,
+    ) -> Optional[Callable[[], Tuple[bytes, bytes]]]:
+        """Return the client cert source to be used by the client."""
+        client_cert_source = None
+        if use_cert_flag:
+            if provided_cert_source:
+                client_cert_source = provided_cert_source
+            elif (
+                hasattr(mtls, "has_default_client_cert_source")
+                and mtls.has_default_client_cert_source()
+            ):
+                client_cert_source = mtls.default_client_cert_source()
+            else:
+                raise ValueError(
+                    "Client certificate is required for mTLS, but no client certificate source was provided or found."
+                )
+        return client_cert_source
+
+    def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
+        """Returns the environment variables used by the client."""
+        use_client_cert = use_client_cert_effective()
+        use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
+        universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
+        if use_mtls_endpoint not in ("auto", "never", "always"):
+            raise MutualTLSChannelError(
+                "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` "
+                "must be `never`, `auto` or `always`"
+            )
+        return use_client_cert, use_mtls_endpoint, universe_domain_env
+
 
 try:
     from google.api_core.gapic_v1.request import setup_request_id  # type: ignore
@@ -40,4 +172,3 @@ except ImportError:
         else:
             if not getattr(request, field_name, None):
                 setattr(request, field_name, request_id_val)
-