⚡️ Speed up function get_bearer_token by 29%

[codeflash-ai]

📄 29% (0.29x) speedup for get_bearer_token in google/cloud/aiplatform/vertex_ray/util/_validation_utils.py

⏱️ Runtime : 656 microseconds → 509 microseconds (best of 545 runs)

📝 Explanation and details

The optimization eliminates the repeated instantiation of google.auth.transport.requests.Request() objects by creating a single global _auth_req instance at module load time and reusing it across all function calls.

Key Changes:

• Moved google.auth.transport.requests.Request() construction from inside the function to module level as _auth_req
• Replaced auth_req = google.auth.transport.requests.Request() with direct usage of the pre-created _auth_req object

Why This Speeds Up Performance:
The line profiler shows that Request() instantiation consumed 16.5% of the original function's runtime (508,216 nanoseconds). By eliminating this repeated object creation, the optimized version removes this overhead entirely. The Request object is stateless and thread-safe, making it safe to reuse across multiple calls.

Test Case Performance:
The optimization delivers consistent 15-40% speedups across all test scenarios, with particularly strong gains in:

• High-frequency calls (32% faster on 500 iterations)
• Many consecutive calls (26% faster on 100 iterations)
• Unicode token handling (39% faster)

This optimization is especially beneficial for applications that frequently request bearer tokens, as it eliminates redundant object allocation on every authentication request.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 1123 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import sys
import types

import google.auth
import google.auth.transport.requests
# imports
import pytest  # used for our unit tests
from aiplatform.vertex_ray.util._validation_utils import get_bearer_token


class DummyCreds:
    """
    Dummy credentials object for simulating google.auth.credentials.
    """
    def __init__(self, token=None, valid=False, refresh_exception=None):
        self.token = token
        self.valid = valid
        self._refresh_exception = refresh_exception
        self._refresh_called = False

    def refresh(self, request):
        self._refresh_called = True
        if self._refresh_exception:
            raise self._refresh_exception
        # Simulate token getting set on refresh
        self.token = self.token or "dummy_token"
        self.valid = True

# ---- Basic Test Cases ----

def test_returns_token_string(monkeypatch):
    """Test that a valid token string is returned under normal conditions."""
    dummy_token = "test_token_123"

    def fake_default(scopes=None):
        return DummyCreds(token=dummy_token, valid=True), None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 2.90μs -> 2.40μs (20.8% faster)

def test_refresh_called_when_token_none(monkeypatch):
    """Test that refresh is called if token is None initially."""
    dummy_creds = DummyCreds(token=None, valid=False)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.59μs -> 1.28μs (24.5% faster)

def test_refresh_sets_token(monkeypatch):
    """Test that refresh sets the token and returns it."""
    dummy_creds = DummyCreds(token=None, valid=False)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.48μs -> 1.26μs (17.4% faster)

# ---- Edge Test Cases ----

def test_refresh_raises_exception(monkeypatch):
    """Test that an exception during refresh propagates."""
    dummy_creds = DummyCreds(token=None, valid=False, refresh_exception=RuntimeError("refresh failed"))

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    with pytest.raises(RuntimeError) as excinfo:
        get_bearer_token() # 1.73μs -> 1.50μs (15.0% faster)

def test_token_is_empty_string(monkeypatch):
    """Test that an empty string token is returned if set by refresh."""
    dummy_creds = DummyCreds(token="", valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.59μs -> 1.34μs (18.8% faster)

def test_token_is_non_ascii(monkeypatch):
    """Test that a non-ASCII token is handled correctly."""
    dummy_token = "токен_非ASCII_令牌"
    dummy_creds = DummyCreds(token=dummy_token, valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.49μs -> 1.27μs (17.3% faster)

def test_token_is_long(monkeypatch):
    """Test that a very long token string is handled."""
    long_token = "A" * 2048
    dummy_creds = DummyCreds(token=long_token, valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.48μs -> 1.22μs (21.7% faster)

def test_creds_object_missing_token_attribute(monkeypatch):
    """Test that AttributeError is raised if creds object lacks 'token'."""
    class NoToken:
        def refresh(self, req): pass

    def fake_default(scopes=None):
        return NoToken(), None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    with pytest.raises(AttributeError):
        get_bearer_token() # 2.35μs -> 2.06μs (14.2% faster)

def test_creds_object_missing_refresh(monkeypatch):
    """Test that AttributeError is raised if creds object lacks 'refresh'."""
    class NoRefresh:
        token = "abc"

    def fake_default(scopes=None):
        return NoRefresh(), None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    with pytest.raises(AttributeError):
        get_bearer_token() # 2.15μs -> 1.80μs (19.4% faster)

def test_google_auth_default_raises(monkeypatch):
    """Test that exceptions from google.auth.default are propagated."""
    def fake_default(scopes=None):
        raise ImportError("No credentials found")

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    with pytest.raises(ImportError) as excinfo:
        get_bearer_token() # 1.28μs -> 1.26μs (1.58% faster)

def test_request_object_is_none(monkeypatch):
    """Test that passing a None request object still works if refresh doesn't use it."""
    dummy_creds = DummyCreds(token="abc", valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return None

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.61μs -> 1.38μs (17.1% faster)

# ---- Large Scale Test Cases ----

def test_many_consecutive_calls(monkeypatch):
    """Test that multiple consecutive calls each return a token and are independent."""
    tokens = ["token_%d" % i for i in range(100)]
    calls = []

    class CountingCreds(DummyCreds):
        def refresh(self, request):
            idx = len(calls)
            self.token = tokens[idx]
            calls.append(idx)
            self.valid = True

    def fake_default(scopes=None):
        return CountingCreds(token=None, valid=False), None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    results = []
    for i in range(100):
        codeflash_output = get_bearer_token(); token = codeflash_output # 83.7μs -> 66.4μs (26.0% faster)
        results.append(token)

def test_large_token_list(monkeypatch):
    """Test that function can handle very large token strings (up to 1000 chars)."""
    large_token = "X" * 1000
    dummy_creds = DummyCreds(token=large_token, valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.50μs -> 1.34μs (12.4% faster)

def test_high_frequency_calls(monkeypatch):
    """Test that function is stable under high-frequency repeated calls."""
    dummy_creds = DummyCreds(token="stable_token", valid=True)

    def fake_default(scopes=None):
        return dummy_creds, None

    def fake_request():
        return "request_obj"

    monkeypatch.setattr(google.auth, "default", fake_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", fake_request)

    for _ in range(500):
        codeflash_output = get_bearer_token(); token = codeflash_output # 243μs -> 184μs (32.0% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import types

import google.auth
import google.auth.transport.requests
# imports
import pytest  # used for our unit tests
from aiplatform.vertex_ray.util._validation_utils import get_bearer_token

# unit tests

@pytest.fixture(autouse=True)
def patch_google_auth(monkeypatch):
    """
    Patch google.auth.default and google.auth.transport.requests.Request for all tests.
    This allows us to simulate various scenarios without external dependencies.
    """
    class DummyCreds:
        def __init__(self, valid=False, token=None, refresh_side_effect=None):
            self.valid = valid
            self.token = token
            self._refresh_side_effect = refresh_side_effect
            self.refresh_called = False

        def refresh(self, request):
            self.refresh_called = True
            if self._refresh_side_effect:
                raise self._refresh_side_effect
            # Simulate setting token after refresh
            self.token = "dummy_token_value"
            self.valid = True

    dummy_creds = DummyCreds()

    def dummy_google_auth_default(scopes=None):
        # Return dummy creds and dummy project
        return dummy_creds, "dummy_project"

    class DummyRequest:
        pass

    monkeypatch.setattr(google.auth, "default", dummy_google_auth_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", DummyRequest)

    # Attach dummy_creds to the test context for access in tests
    yield dummy_creds

#############
# Basic Test Cases
#############

def test_returns_token_string(patch_google_auth):
    """Test that get_bearer_token returns a non-empty string token under normal conditions."""
    patch_google_auth.token = None  # Ensure token is None before refresh
    codeflash_output = get_bearer_token(); token = codeflash_output # 1.70μs -> 1.32μs (28.2% faster)

def test_token_is_refreshed_if_invalid(patch_google_auth):
    """Test that get_bearer_token refreshes token if creds are invalid."""
    patch_google_auth.valid = False
    patch_google_auth.token = None
    codeflash_output = get_bearer_token(); token = codeflash_output # 1.59μs -> 1.18μs (34.5% faster)

def test_token_is_refreshed_even_if_already_valid(patch_google_auth):
    """Test that get_bearer_token always refreshes token (as per implementation)."""
    patch_google_auth.valid = True
    patch_google_auth.token = "old_token"
    codeflash_output = get_bearer_token(); token = codeflash_output # 1.58μs -> 1.17μs (34.5% faster)

#############
# Edge Test Cases
#############

def test_refresh_raises_exception(patch_google_auth):
    """Test that get_bearer_token propagates exceptions from refresh()."""
    class DummyException(Exception):
        pass
    patch_google_auth._refresh_side_effect = DummyException("refresh failed")
    patch_google_auth.token = None
    with pytest.raises(DummyException):
        get_bearer_token() # 1.78μs -> 1.56μs (14.2% faster)
    patch_google_auth._refresh_side_effect = None  # Reset for other tests

def test_token_is_none_after_refresh(patch_google_auth):
    """Test behavior if token remains None after refresh (should return None)."""
    # Patch refresh to NOT set token
    def no_token_refresh(request):
        patch_google_auth.refresh_called = True
        patch_google_auth.token = None
        patch_google_auth.valid = True
    patch_google_auth.refresh = no_token_refresh
    patch_google_auth.token = None
    codeflash_output = get_bearer_token(); token = codeflash_output # 1.57μs -> 1.23μs (27.8% faster)

def test_refresh_called_with_request_object(patch_google_auth):
    """Test that refresh() is called with a Request object."""
    called_args = {}
    def refresh(request):
        called_args['request'] = request
        patch_google_auth.token = "dummy_token_value"
        patch_google_auth.valid = True
    patch_google_auth.refresh = refresh
    codeflash_output = get_bearer_token(); token = codeflash_output # 1.49μs -> 1.18μs (26.4% faster)

def test_scopes_are_passed_to_default(monkeypatch, patch_google_auth):
    """Test that the correct scopes are passed to google.auth.default."""
    scopes_passed = {}
    def dummy_google_auth_default(scopes=None):
        scopes_passed['scopes'] = scopes
        return patch_google_auth, "dummy_project"
    monkeypatch.setattr(google.auth, "default", dummy_google_auth_default)
    get_bearer_token() # 1.79μs -> 1.41μs (27.5% faster)

def test_multiple_calls_return_new_token_each_time(patch_google_auth):
    """Test that each call to get_bearer_token returns a fresh token."""
    call_count = {'count': 0}
    def refresh(request):
        call_count['count'] += 1
        patch_google_auth.token = f"token_{call_count['count']}"
        patch_google_auth.valid = True
    patch_google_auth.refresh = refresh
    codeflash_output = get_bearer_token(); token1 = codeflash_output # 1.96μs -> 1.63μs (20.1% faster)
    codeflash_output = get_bearer_token(); token2 = codeflash_output # 904ns -> 738ns (22.5% faster)

#############
# Large Scale Test Cases
#############

def test_many_parallel_token_requests(monkeypatch):
    """Test get_bearer_token under many simulated concurrent requests."""
    # We'll simulate 500 parallel requests (well under 1000 as per instructions)
    # Each will get a unique token
    class DummyCreds:
        def __init__(self):
            self.valid = False
            self.token = None
            self.refresh_called = False
            self.counter = 0

        def refresh(self, request):
            self.refresh_called = True
            self.counter += 1
            self.token = f"token_{self.counter}"

    dummy_creds = DummyCreds()

    def dummy_google_auth_default(scopes=None):
        return dummy_creds, "dummy_project"

    class DummyRequest:
        pass

    monkeypatch.setattr(google.auth, "default", dummy_google_auth_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", DummyRequest)

    tokens = []
    for _ in range(500):
        dummy_creds.valid = False
        dummy_creds.token = None
        codeflash_output = get_bearer_token(); token = codeflash_output # 290μs -> 226μs (28.1% faster)
        tokens.append(token)

def test_token_length_and_content_large_token(monkeypatch):
    """Test get_bearer_token with a very large token value."""
    # Simulate a token of length 1024 characters
    class DummyCreds:
        def __init__(self):
            self.valid = False
            self.token = None

        def refresh(self, request):
            self.token = "X" * 1024
            self.valid = True

    dummy_creds = DummyCreds()
    def dummy_google_auth_default(scopes=None):
        return dummy_creds, "dummy_project"
    class DummyRequest:
        pass
    monkeypatch.setattr(google.auth, "default", dummy_google_auth_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", DummyRequest)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.32μs -> 1.10μs (20.2% faster)

def test_token_is_unicode(monkeypatch):
    """Test that get_bearer_token can handle unicode tokens."""
    class DummyCreds:
        def __init__(self):
            self.valid = False
            self.token = None

        def refresh(self, request):
            self.token = "токен_значение_测试"
            self.valid = True

    dummy_creds = DummyCreds()
    def dummy_google_auth_default(scopes=None):
        return dummy_creds, "dummy_project"
    class DummyRequest:
        pass
    monkeypatch.setattr(google.auth, "default", dummy_google_auth_default)
    monkeypatch.setattr(google.auth.transport.requests, "Request", DummyRequest)

    codeflash_output = get_bearer_token(); token = codeflash_output # 1.38μs -> 992ns (39.3% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-get_bearer_token-mglrv9o1 and push.