⚡️ Speed up function fetch_all_users by 900%

[codeflash-ai]

📄 900% (9.00x) speedup for fetch_all_users in src/asynchrony/various.py

⏱️ Runtime : 7.38 seconds → 1.09 seconds (best of 70 runs)

📝 Explanation and details

The optimization replaces sequential async execution with concurrent execution using asyncio.gather(), delivering a dramatic 574% speedup and 900% throughput improvement.

Key Change:

• Original: Sequential loop awaiting each fetch_user(user_id) one at a time
• Optimized: Single line using asyncio.gather(*(fetch_user(user_id) for user_id in user_ids)) to execute all calls concurrently

Why This Works:
The original code's sequential approach means each fetch_user call (with its 0.001s sleep) blocks the next one. For N users, total time = N × 0.001s. The optimized version launches all fetch_user tasks simultaneously, so total time ≈ 0.001s regardless of user count, since all I/O operations happen in parallel.

Performance Impact:

• Runtime: From 7.38s to 1.09s - the concurrent execution eliminates the cumulative waiting time
• Throughput: From 1,141 to 11,410 operations/second - can process ~10x more requests per unit time
• Line profiler confirms: The optimized version spends almost all time (100%) in the single asyncio.gather() call rather than looping through individual awaits

Workload Benefits:
This optimization is particularly effective for:

• Large user lists (test cases with 100-500 users show the most benefit)
• High-volume concurrent scenarios (throughput tests with 50+ concurrent calls)
• Any I/O-bound batch operations where individual tasks are independent

The optimization maintains identical behavior, order preservation, and error handling while maximizing async concurrency benefits.

✅ Correctness verification report:

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

⚙️ Existing Unit Tests and Runtime

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions

import pytest  # used for our unit tests
from src.asynchrony.various import fetch_all_users

# unit tests

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_empty_list():
    # Test with an empty list; should return an empty list
    result = await fetch_all_users([])

@pytest.mark.asyncio
async def test_fetch_all_users_single_user():
    # Test with a single user ID
    result = await fetch_all_users([42])

@pytest.mark.asyncio
async def test_fetch_all_users_multiple_users():
    # Test with multiple user IDs
    user_ids = [1, 2, 3]
    expected = [
        {"id": 1, "name": "User1"},
        {"id": 2, "name": "User2"},
        {"id": 3, "name": "User3"}
    ]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_basic_async_behavior():
    # Test that function returns a coroutine and can be awaited
    codeflash_output = fetch_all_users([5]); coro = codeflash_output
    result = await coro

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_duplicate_ids():
    # Test with duplicate user IDs; should return duplicate user dicts
    user_ids = [7, 7, 7]
    expected = [
        {"id": 7, "name": "User7"},
        {"id": 7, "name": "User7"},
        {"id": 7, "name": "User7"}
    ]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_negative_and_zero_ids():
    # Test with negative and zero user IDs
    user_ids = [-1, 0, 1]
    expected = [
        {"id": -1, "name": "User-1"},
        {"id": 0, "name": "User0"},
        {"id": 1, "name": "User1"}
    ]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_non_sequential_ids():
    # Test with non-sequential, unordered IDs
    user_ids = [100, 3, 50]
    expected = [
        {"id": 100, "name": "User100"},
        {"id": 3, "name": "User3"},
        {"id": 50, "name": "User50"}
    ]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_concurrent_calls():
    # Test concurrent execution of fetch_all_users with different inputs
    ids1 = [1, 2]
    ids2 = [3, 4]
    expected1 = [{"id": 1, "name": "User1"}, {"id": 2, "name": "User2"}]
    expected2 = [{"id": 3, "name": "User3"}, {"id": 4, "name": "User4"}]
    results = await asyncio.gather(
        fetch_all_users(ids1),
        fetch_all_users(ids2)
    )

@pytest.mark.asyncio


async def test_fetch_all_users_large_list():
    # Test with a large list of user IDs (but <1000 for speed)
    user_ids = list(range(100))
    expected = [{"id": i, "name": f"User{i}"} for i in user_ids]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_large_concurrent_calls():
    # Test multiple concurrent fetch_all_users calls with large lists
    ids_list = [list(range(i, i+50)) for i in range(0, 200, 50)]
    expected_list = [[{"id": id_, "name": f"User{id_}"} for id_ in ids] for ids in ids_list]
    results = await asyncio.gather(*(fetch_all_users(ids) for ids in ids_list))
    for result, expected in zip(results, expected_list):
        pass

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_small_load():
    # Throughput test: small load, repeated calls
    user_ids = [1, 2, 3]
    for _ in range(10):
        result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_medium_load():
    # Throughput test: medium load, concurrent calls
    user_ids = list(range(20))
    tasks = [fetch_all_users(user_ids) for _ in range(10)]
    results = await asyncio.gather(*tasks)
    expected = [{"id": i, "name": f"User{i}"} for i in user_ids]
    for result in results:
        pass

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_high_volume():
    # Throughput test: high volume, many concurrent calls with larger lists
    user_ids = list(range(100))
    tasks = [fetch_all_users(user_ids) for _ in range(20)]
    results = await asyncio.gather(*tasks)
    expected = [{"id": i, "name": f"User{i}"} for i in user_ids]
    for result in results:
        pass

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_varied_input_sizes():
    # Throughput test: varied input sizes in concurrent calls
    input_sizes = [0, 1, 10, 50, 100]
    tasks = [fetch_all_users(list(range(size))) for size in input_sizes]
    results = await asyncio.gather(*tasks)
    for size, result in zip(input_sizes, results):
        expected = [{"id": i, "name": f"User{i}"} for i in range(size)]
# 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

import pytest  # used for our unit tests
from src.asynchrony.various import fetch_all_users

# -----------------
# unit tests

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_empty_list():
    """Test with an empty list of user_ids."""
    result = await fetch_all_users([])

@pytest.mark.asyncio
async def test_fetch_all_users_single_user():
    """Test with a single user_id."""
    result = await fetch_all_users([42])

@pytest.mark.asyncio
async def test_fetch_all_users_multiple_users():
    """Test with multiple user_ids."""
    user_ids = [1, 2, 3]
    result = await fetch_all_users(user_ids)
    for i, user in enumerate(result):
        pass

@pytest.mark.asyncio
async def test_fetch_all_users_duplicate_ids():
    """Test with duplicate user IDs in input."""
    user_ids = [5, 5, 7]
    result = await fetch_all_users(user_ids)

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_negative_and_zero_ids():
    """Test with zero and negative user IDs."""
    user_ids = [0, -1, -999]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio

async def test_fetch_all_users_concurrent_execution():
    """Test concurrent calls to fetch_all_users with different input lists."""
    user_ids_1 = [1, 2]
    user_ids_2 = [3, 4]
    # Run two fetch_all_users concurrently
    results = await asyncio.gather(
        fetch_all_users(user_ids_1),
        fetch_all_users(user_ids_2),
    )

@pytest.mark.asyncio
async def test_fetch_all_users_large_id_values():
    """Test with very large integer user IDs."""
    user_ids = [2**30, 2**31-1]
    result = await fetch_all_users(user_ids)

@pytest.mark.asyncio
async def test_fetch_all_users_order_preservation():
    """Test that output order matches input order."""
    user_ids = [10, 5, 20]
    result = await fetch_all_users(user_ids)

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_large_list():
    """Test with a large list of user IDs (up to 500)."""
    user_ids = list(range(500))
    result = await fetch_all_users(user_ids)
    for i in range(500):
        pass

@pytest.mark.asyncio
async def test_fetch_all_users_concurrent_large_lists():
    """Test concurrent execution with multiple large lists."""
    user_ids_1 = list(range(100, 200))
    user_ids_2 = list(range(200, 300))
    results = await asyncio.gather(
        fetch_all_users(user_ids_1),
        fetch_all_users(user_ids_2),
    )

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_small_load():
    """Throughput test with a small number of concurrent calls."""
    user_ids = [1, 2, 3]
    coros = [fetch_all_users(user_ids) for _ in range(10)]
    results = await asyncio.gather(*coros)
    for result in results:
        pass

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_medium_load():
    """Throughput test with a medium number of concurrent calls."""
    user_ids = list(range(20))
    coros = [fetch_all_users(user_ids) for _ in range(20)]
    results = await asyncio.gather(*coros)
    for result in results:
        for i in range(20):
            pass

@pytest.mark.asyncio
async def test_fetch_all_users_throughput_high_volume():
    """Throughput test with high volume: many concurrent calls with moderate input size."""
    user_ids = list(range(50))
    coros = [fetch_all_users(user_ids) for _ in range(50)]
    results = await asyncio.gather(*coros)
    for result in results:
        for i in range(50):
            pass
# 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-fetch_all_users-mhv8y6h6 and push.