feat(auth): PEP 0810 explicit lazy imports in google-auth transports

This PR implements PEP 0810 explicit lazy imports in google-auth transports (requests and urllib3).
On Python 3.15+, this defers loading of heavy third-party networking libraries (requests and urllib3) to reduce serverless cold-start latency and peak memory usage. On Python 3.14 and below, this falls back safely to standard eager loading with zero backwards-compatibility risk.

Author: hebaalazzeh

packages/google-auth/google/auth/_helpers.py

@@ -22,6 +22,7 @@
 import json
 import logging
 import sys
+import threading
 from typing import Any, Dict, Mapping, Optional, Union
 import urllib
 
@@ -534,3 +535,69 @@ def response_log(logger: logging.Logger, response: Any) -> None:
     if is_logging_enabled(logger):
         json_response = _parse_response(response)
         _response_log_base(logger, json_response)
+
+
+class LazyBasesMeta(type):
+    """Metaclass that allows dynamic lazy resolution of class bases.
+
+    This metaclass avoids eager loading of heavy dependencies during import
+    time by allowing classes to inherit from a dummy class initially.
+    The real base classes are dynamically resolved and swapped in when the
+    class is first instantiated, subclassed, or has its attributes inspected.
+    """
+
+    _lock = threading.RLock()
+
+    def __new__(mcls, name, bases, attrs):
+        # Automatically resolve lazy bases of any base classes at subclass definition time.
+        for base in bases:
+            if isinstance(base, LazyBasesMeta):
+                type(base)._resolve_bases(base)
+
+        cls = super(LazyBasesMeta, mcls).__new__(mcls, name, bases, attrs)
+        type.__setattr__(cls, "_lazy_bases_resolved", False)
+        return cls
+
+    def __call__(cls, *args, **kwargs):
+        type(cls)._resolve_bases(cls)
+        return super(LazyBasesMeta, cls).__call__(*args, **kwargs)
+
+    def __getattribute__(cls, name):
+        if name not in (
+            "_lazy_bases_resolved",
+            "_resolve_bases",
+            "_perform_resolve_bases",
+        ):
+            type(cls)._resolve_bases(cls)
+        return super(LazyBasesMeta, cls).__getattribute__(name)
+
+    def _resolve_bases(cls):
+        try:
+            resolved = type.__getattribute__(cls, "_lazy_bases_resolved")
+        except AttributeError:
+            resolved = False
+
+        if not resolved:
+            with type(cls)._lock:
+                try:
+                    resolved = type.__getattribute__(cls, "_lazy_bases_resolved")
+                except AttributeError:
+                    resolved = False
+
+                if not resolved:
+                    type(cls)._perform_resolve_bases(cls)
+                    type.__setattr__(cls, "_lazy_bases_resolved", True)
+
+    def _perform_resolve_bases(cls):
+        pass
+
+
+class HeapDummy(object):
+    """A heap-allocated dummy class.
+
+    This class serves as a safe initial base class for classes using LazyBasesMeta.
+    Inheriting from HeapDummy instead of built-in `object` ensures that Python's
+    memory layout check passes when we swap bases to another heap-allocated class.
+    """
+
+    pass

packages/google-auth/google/auth/transport/requests.py

@@ -18,30 +18,63 @@
 
 import functools
 import http.client as http_client
+import importlib.util
 import logging
 import numbers
 import time
-from typing import Optional
-
-try:
-    import requests
-except ImportError as caught_exc:  # pragma: NO COVER
+from typing import Optional, Set, TYPE_CHECKING
+
+# PEP 0810: Explicit Lazy Imports
+# Python 3.15+ natively intercepts and defers these imports.
+# Developers can disable this behavior and force eager imports.
+# For more information, see:
+# https://docs.python.org/3.15/library/sys.html#sys.set_lazy_imports_filter
+# Older Python versions safely ignore this variable.
+__lazy_modules__: Set[str] = {
+    "requests",
+    "requests.adapters",
+    "requests.exceptions",
+    "requests.packages",
+    "requests.packages.urllib3",
+    "requests.packages.urllib3.util",
+    "requests.packages.urllib3.util.ssl_",
+}
+
+
+if importlib.util.find_spec("requests") is None:
     raise ImportError(
-        "The requests library is not installed from please install the requests package to use the requests transport."
-    ) from caught_exc
+        "The requests library is not installed, please install the requests package to use the requests transport."
+    )
+
+import requests
 import requests.adapters  # pylint: disable=ungrouped-imports
 import requests.exceptions  # pylint: disable=ungrouped-imports
-from requests.packages.urllib3.util.ssl_ import (  # type: ignore
-    create_urllib3_context,
-)  # pylint: disable=ungrouped-imports
 
-from google.auth import _helpers
-from google.auth import exceptions
-from google.auth import transport
+
+from google.auth import _helpers, exceptions, transport
 from google.auth.transport import _mtls_helper
 import google.auth.transport._mtls_helper
 from google.oauth2 import service_account
 
+if TYPE_CHECKING:
+    _HTTPAdapterBase = requests.adapters.HTTPAdapter
+    _SessionBase = requests.Session
+else:
+    _HTTPAdapterBase = _helpers.HeapDummy
+    _SessionBase = _helpers.HeapDummy
+
+
+class _LazyBasesMeta(_helpers.LazyBasesMeta):
+    def _perform_resolve_bases(cls):
+        current_bases = type.__getattribute__(cls, "__bases__")
+        if current_bases == (_helpers.HeapDummy,):
+            cls_name = type.__getattribute__(cls, "__name__")
+            if cls_name in ("_MutualTlsAdapter", "_MutualTlsOffloadAdapter"):
+                cls.__bases__ = (requests.adapters.HTTPAdapter,)
+            elif cls_name == "AuthorizedSession":
+                cls.__bases__ = (requests.Session,)
+
+
 _LOGGER = logging.getLogger(__name__)
 
 _DEFAULT_TIMEOUT = 120  # in seconds
@@ -84,9 +117,11 @@ class TimeoutGuard(object):
             :class:`requests.exceptions.Timeout`.
     """
 
-    def __init__(self, timeout, timeout_error_type=requests.exceptions.Timeout):
+    def __init__(self, timeout, timeout_error_type=None):
         self._timeout = timeout
         self.remaining_timeout = timeout
+        if timeout_error_type is None:
+            timeout_error_type = requests.exceptions.Timeout
         self._timeout_error_type = timeout_error_type
 
     def __enter__(self):
@@ -161,7 +196,7 @@ def __call__(
         body=None,
         headers=None,
         timeout=_DEFAULT_TIMEOUT,
-        **kwargs
+        **kwargs,
     ):
         """Make an HTTP request using requests.
 
@@ -195,7 +230,7 @@ def __call__(
             raise new_exc from caught_exc
 
 
-class _MutualTlsAdapter(requests.adapters.HTTPAdapter):
+class _MutualTlsAdapter(_HTTPAdapterBase, metaclass=_LazyBasesMeta):
     """
     A TransportAdapter that enables mutual TLS.
 
@@ -209,9 +244,13 @@ class _MutualTlsAdapter(requests.adapters.HTTPAdapter):
     """
 
     def __init__(self, cert, key):
-        import certifi
         import ssl
 
+        import certifi
+        from requests.packages.urllib3.util.ssl_ import (  # type: ignore
+            create_urllib3_context,
+        )
+
         ctx_poolmanager = create_urllib3_context()
         ctx_poolmanager.load_verify_locations(cafile=certifi.where())
 
@@ -261,7 +300,7 @@ def proxy_manager_for(self, *args, **kwargs):
         return super(_MutualTlsAdapter, self).proxy_manager_for(*args, **kwargs)
 
 
-class _MutualTlsOffloadAdapter(requests.adapters.HTTPAdapter):
+class _MutualTlsOffloadAdapter(_HTTPAdapterBase, metaclass=_LazyBasesMeta):
     """
     A TransportAdapter that enables mutual TLS and offloads the client side
     signing operation to the signing library.
@@ -285,6 +324,10 @@ class _MutualTlsOffloadAdapter(requests.adapters.HTTPAdapter):
 
     def __init__(self, enterprise_cert_file_path):
         import certifi
+        from requests.packages.urllib3.util.ssl_ import (  # type: ignore
+            create_urllib3_context,
+        )
+
         from google.auth.transport import _custom_tls_signer
 
         self.signer = _custom_tls_signer.CustomTlsSigner(enterprise_cert_file_path)
@@ -311,7 +354,7 @@ def proxy_manager_for(self, *args, **kwargs):
         return super(_MutualTlsOffloadAdapter, self).proxy_manager_for(*args, **kwargs)
 
 
-class AuthorizedSession(requests.Session):
+class AuthorizedSession(_SessionBase, metaclass=_LazyBasesMeta):
     """A Requests Session class with credentials.
 
     This class is used to perform requests to API endpoints that require
@@ -504,7 +547,7 @@ def request(
         headers=None,
         max_allowed_time=None,
         timeout=_DEFAULT_TIMEOUT,
-        **kwargs
+        **kwargs,
     ):
         """Implementation of Requests' request.
 
@@ -566,7 +609,7 @@ def request(
                 data=data,
                 headers=request_headers,
                 timeout=timeout,
-                **kwargs
+                **kwargs,
             )
         remaining_time = guard.remaining_timeout
 
@@ -638,7 +681,7 @@ def request(
                 max_allowed_time=remaining_time,
                 timeout=timeout,
                 _credential_refresh_attempt=_credential_refresh_attempt + 1,
-                **kwargs
+                **kwargs,
             )
 
         return response

packages/google-auth/google/auth/transport/urllib3.py

@@ -18,6 +18,7 @@
 
 import http.client as http_client
 import logging
+from typing import Set, TYPE_CHECKING
 import warnings
 
 # Certifi is Mozilla's certificate bundle. Urllib3 needs a certificate bundle
@@ -50,16 +51,47 @@
     ) from caught_exc
 
 
-from google.auth import _helpers
-from google.auth import exceptions
-from google.auth import transport
+from google.auth import _helpers, exceptions, transport
 from google.auth.transport import _mtls_helper
 from google.oauth2 import service_account
 
-if version.parse(urllib3.__version__) >= version.parse("2.0.0"):  # pragma: NO COVER
-    RequestMethods = urllib3._request_methods.RequestMethods  # type: ignore
-else:  # pragma: NO COVER
-    RequestMethods = urllib3.request.RequestMethods  # type: ignore
+if TYPE_CHECKING:
+    try:
+        from urllib3._request_methods import RequestMethods as _HttpBase
+    except ImportError:
+        try:
+            from urllib3.request import RequestMethods as _HttpBase  # type: ignore
+        except ImportError:
+            _HttpBase = object  # type: ignore
+else:
+    _HttpBase = _helpers.HeapDummy
+
+
+# PEP 0810: Explicit Lazy Imports
+# Python 3.15+ natively intercepts and defers these imports.
+# Developers can disable this behavior and force eager imports.
+# For more information, see:
+# https://docs.python.org/3.15/library/sys.html#sys.set_lazy_imports_filter
+# Older Python versions safely ignore this variable.
+__lazy_modules__: Set[str] = {
+    "urllib3",
+    "urllib3.exceptions",
+    "certifi",
+    "packaging",
+    "packaging.version",
+}
+
+
+class _LazyBasesMeta(_helpers.LazyBasesMeta):
+    def _perform_resolve_bases(cls):
+        current_bases = type.__getattribute__(cls, "__bases__")
+        if current_bases == (_helpers.HeapDummy,):
+            if version.parse(urllib3.__version__) >= version.parse("2.0.0"):
+                request_methods_class = urllib3._request_methods.RequestMethods  # type: ignore
+            else:
+                request_methods_class = urllib3.request.RequestMethods  # type: ignore
+            cls.__bases__ = (request_methods_class,)
+
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -176,9 +208,10 @@ def _make_mutual_tls_http(cert, key):
     Raises:
         google.auth.exceptions.MutualTLSChannelError: If the cert or key is invalid.
     """
-    import certifi
     import ssl
 
+    import certifi
+
     ctx = urllib3.util.ssl_.create_urllib3_context()
     ctx.load_verify_locations(cafile=certifi.where())
 
@@ -203,7 +236,7 @@ def _make_mutual_tls_http(cert, key):
     return http
 
 
-class AuthorizedHttp(RequestMethods):  # type: ignore
+class AuthorizedHttp(_HttpBase, metaclass=_LazyBasesMeta):
     """A urllib3 HTTP class with credentials.
 
     This class is used to perform requests to API endpoints that require

packages/google-auth/tests/transport/test_requests.py

@@ -498,7 +498,7 @@ def test_configure_mtls_channel_exceptions(self, mock_get_client_cert_and_key):
     @mock.patch(
         "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True
     )
-    @mock.patch("google.auth.transport.requests.create_urllib3_context", autospec=True)
+    @mock.patch("requests.packages.urllib3.util.ssl_.create_urllib3_context", autospec=True)
     def test_configure_mtls_channel_cert_loading_exceptions(
         self, mock_create_urllib3_context, mock_get_client_cert_and_key
     ):