⚡️ Speed up function get_current_user_id_with_authentication by 22%

[codeflash-ai]

📄 22% (0.22x) speedup for get_current_user_id_with_authentication in skyvern/forge/sdk/services/org_auth_service.py

⏱️ Runtime : 181 microseconds → 149 microseconds (best of 77 runs)

📝 Explanation and details

The optimization inlines the _authenticate_user_helper function directly into get_current_user_id_with_authentication, eliminating an unnecessary async function call layer.

Key changes:

• Removed helper function: The _authenticate_user_helper function is eliminated entirely
• Direct implementation: Token extraction and authentication logic moved directly into the main function
• Reduced async overhead: Eliminates one await call and associated function call overhead

Why this improves performance:
The original code required two async function calls: first get_current_user_id_with_authentication awaiting _authenticate_user_helper, then _authenticate_user_helper awaiting app.authenticate_user_function. The optimization reduces this to a single async chain by inlining the helper logic.

From the profiler data, the original code spent 96.3% of time (10.26ms out of 10.66ms total) just on the await _authenticate_user_helper(authorization) call. The optimized version eliminates this intermediate async call, reducing total execution time from 181 to 149 microseconds.

Performance impact:

• 21% runtime improvement: From 181μs to 149μs
• 8.5% throughput increase: From 165,501 to 179,487 operations/second
• Reduced function call overhead: Eliminates async function dispatch and context switching

This optimization is particularly effective for high-frequency authentication scenarios where the function is called repeatedly, as each request saves the overhead of an additional async function call. The test cases show the optimization benefits all scenarios - from basic authentication failures to concurrent request handling - by reducing the baseline execution cost for every invocation.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# Patch the global app object in the module namespace
import sys
# --- Function under test (copied EXACTLY as provided) ---
from typing import Annotated

import pytest  # used for our unit tests
from fastapi import Header, HTTPException, status
from skyvern.forge.sdk.services.org_auth_service import \
    get_current_user_id_with_authentication

# Simulate the skyvern.forge.app global object
class DummyApp:
    authenticate_user_function = None

# Patch the global app for testing
class DummyAppWithAuth(DummyApp):
    def __init__(self, return_value=None, raise_exc=False, delay=0):
        async def auth_func(token):
            if delay:
                await asyncio.sleep(delay)
            if raise_exc:
                raise Exception("Unexpected error")
            return return_value
        self.authenticate_user_function = auth_func

module = sys.modules[__name__]
from skyvern.forge.sdk.services.org_auth_service import \
    get_current_user_id_with_authentication

# --- UNIT TESTS ---

@pytest.mark.asyncio

async def test_basic_missing_authorization_header_raises():
    """Test missing authorization header raises HTTPException."""
    module.app = DummyAppWithAuth(return_value="user123")
    with pytest.raises(HTTPException) as exc_info:
        await get_current_user_id_with_authentication(None)

@pytest.mark.asyncio
async def test_basic_invalid_token_returns_raises():
    """Test invalid token (authenticate_user_function returns None) raises."""
    module.app = DummyAppWithAuth(return_value=None)
    with pytest.raises(HTTPException) as exc_info:
        await get_current_user_id_with_authentication("Bearer invalidtoken")

@pytest.mark.asyncio
async def test_basic_authenticate_user_function_not_set_raises():
    """Test when authenticate_user_function is not set, raises HTTPException."""
    module.app = DummyApp()
    with pytest.raises(HTTPException) as exc_info:
        await get_current_user_id_with_authentication("Bearer sometoken")

@pytest.mark.asyncio

async def test_edge_authorization_header_wrong_format():
    """Test authorization header with wrong format (missing token) raises IndexError."""
    module.app = DummyAppWithAuth(return_value="user789")
    # This should raise IndexError because split()[1] will fail
    with pytest.raises(IndexError):
        await get_current_user_id_with_authentication("Bearer")

@pytest.mark.asyncio
async def test_edge_authenticate_user_function_raises_exception():
    """Test if authenticate_user_function raises an unexpected exception."""
    module.app = DummyAppWithAuth(return_value="user999", raise_exc=True)
    with pytest.raises(Exception) as exc_info:
        await get_current_user_id_with_authentication("Bearer sometoken")

@pytest.mark.asyncio

async def test_edge_concurrent_requests_with_mixed_results():
    """Test concurrent requests with mixed valid and invalid tokens."""
    # Valid returns userX, invalid returns None
    class MixedDummyApp(DummyApp):
        async def authenticate_user_function(self, token):
            if token == "validtoken":
                return "userX"
            return None
    module.app = MixedDummyApp()
    tokens = [
        "Bearer validtoken",  # valid
        "Bearer invalidtoken",  # invalid
        "Bearer validtoken",  # valid
        "Bearer invalidtoken",  # invalid
    ]
    coros = [get_current_user_id_with_authentication(token) for token in tokens]
    # Run concurrently, catching exceptions
    results = []
    for coro in coros:
        try:
            result = await coro
            results.append(result)
        except HTTPException as e:
            results.append(e.detail)

@pytest.mark.asyncio

async def test_large_scale_many_concurrent_requests_with_some_failures():
    """Test large scale: concurrent requests, some fail, some succeed."""
    class PartialFailDummyApp(DummyApp):
        async def authenticate_user_function(self, token):
            # Fail every 10th token
            if token.endswith("9"):
                return None
            return "user_bulk"
    module.app = PartialFailDummyApp()
    tokens = [f"Bearer token{i}" for i in range(50)]
    coros = [get_current_user_id_with_authentication(token) for token in tokens]
    results = []
    for coro, i in zip(coros, range(50)):
        try:
            result = await coro
            results.append(result)
        except HTTPException as e:
            results.append(e.detail)
    for i, res in enumerate(results):
        if i % 10 == 9:
            pass
        else:
            pass

@pytest.mark.asyncio

async def test_get_current_user_id_with_authentication_throughput_mixed_load():
    """Throughput test: mixed load, some requests fail."""
    class MixedThroughputApp(DummyApp):
        async def authenticate_user_function(self, token):
            # Fail every 5th token
            if int(token.replace("token", "")) % 5 == 0:
                return None
            return "user_throughput"
    module.app = MixedThroughputApp()
    tokens = [f"Bearer token{i}" for i in range(30)]
    coros = [get_current_user_id_with_authentication(token) for token in tokens]
    results = []
    for i, coro in enumerate(coros):
        try:
            result = await coro
            results.append(result)
        except HTTPException as e:
            results.append(e.detail)
    for i, res in enumerate(results):
        if i % 5 == 0:
            pass
        else:
            pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import asyncio  # used to run async functions
# Patch the global app used by the function
import sys
# --- Test Setup: Patch app.authenticate_user_function ---
import types
from typing import Annotated

import pytest  # used for our unit tests
# --- Function to test (copied EXACTLY as provided) ---
from fastapi import Header, HTTPException, status
from skyvern.forge import app
from skyvern.forge.sdk.services.org_auth_service import \
    get_current_user_id_with_authentication

class DummyApp:
    authenticate_user_function = None

sys.modules['skyvern.forge'].app = DummyApp()

# Helper: Set up dummy authenticate_user_function for tests
def set_auth_func(func):
    sys.modules['skyvern.forge'].app.authenticate_user_function = func

# --- Basic Test Cases ---

@pytest.mark.asyncio

async def test_raises_for_missing_authorization():
    # Setup: any auth function
    async def dummy_auth(token): return "user123"
    set_auth_func(dummy_auth)
    # Call function with None header
    with pytest.raises(HTTPException) as excinfo:
        await get_current_user_id_with_authentication(None)

@pytest.mark.asyncio
async def test_raises_for_empty_authorization():
    # Setup: any auth function
    async def dummy_auth(token): return "user123"
    set_auth_func(dummy_auth)
    # Call function with empty string header
    with pytest.raises(HTTPException) as excinfo:
        await get_current_user_id_with_authentication("")

@pytest.mark.asyncio
async def test_raises_for_missing_authenticate_user_function():
    # Setup: authenticate_user_function is None
    set_auth_func(None)
    # Call with valid header
    with pytest.raises(HTTPException) as excinfo:
        await get_current_user_id_with_authentication("Bearer validtoken")

@pytest.mark.asyncio
async def test_raises_for_invalid_token():
    # Setup: auth function returns None for invalid token
    async def dummy_auth(token):
        return None
    set_auth_func(dummy_auth)
    # Call with invalid token
    with pytest.raises(HTTPException) as excinfo:
        await get_current_user_id_with_authentication("Bearer badtoken")

@pytest.mark.asyncio
async def test_raises_for_auth_function_returning_empty_string():
    # Setup: auth function returns empty string
    async def dummy_auth(token):
        return ""
    set_auth_func(dummy_auth)
    # Call with valid token
    with pytest.raises(HTTPException) as excinfo:
        await get_current_user_id_with_authentication("Bearer validtoken")

# --- Edge Test Cases ---

@pytest.mark.asyncio
async def test_raises_for_authorization_without_bearer_prefix():
    # Setup: valid auth function
    async def dummy_auth(token):
        return "user123"
    set_auth_func(dummy_auth)
    # Call with header missing 'Bearer' prefix
    with pytest.raises(IndexError):
        # This will cause split(" ")[1] to fail
        await get_current_user_id_with_authentication("InvalidHeaderOnlyToken")

@pytest.mark.asyncio

async def test_concurrent_missing_authenticate_user_function():
    # Setup: auth function is None
    set_auth_func(None)
    # Prepare multiple calls
    async def call():
        with pytest.raises(HTTPException) as excinfo:
            await get_current_user_id_with_authentication("Bearer validtoken")
    await asyncio.gather(call(), call())

@pytest.mark.asyncio

async def test_many_concurrent_invalid_tokens():
    # Setup: auth function returns None for all tokens
    async def dummy_auth(token):
        return None
    set_auth_func(dummy_auth)
    # Prepare 50 concurrent calls
    headers = [f"Bearer badtoken{i}" for i in range(50)]
    coros = [
        get_current_user_id_with_authentication(header)
        for header in headers
    ]
    results = []
    for coro in coros:
        try:
            await coro
        except HTTPException as exc:
            results.append("exception")

@pytest.mark.asyncio
async def test_mixed_concurrent_valid_and_invalid_tokens():
    # Setup: auth function returns user_id for even, None for odd
    async def dummy_auth(token):
        idx = int(token.replace("token", ""))
        return f"user_{token}" if idx % 2 == 0 else None
    set_auth_func(dummy_auth)
    # Prepare 20 concurrent calls
    headers = [f"Bearer token{i}" for i in range(20)]
    coros = [
        get_current_user_id_with_authentication(header)
        for header in headers
    ]
    results = []
    for i, coro in enumerate(coros):
        try:
            res = await coro
            results.append("ok")
        except HTTPException as exc:
            results.append("exception")

# --- Throughput Test Cases ---

@pytest.mark.asyncio

To edit these changes git checkout codeflash/optimize-get_current_user_id_with_authentication-mjampodf and push.