L2 · Working engineerAdvanced L2~12 min · 3 tests#92

Make Concurrent API Calls With asyncio

Fetch five APIs concurrently with asyncio.gather so total time is the slowest call, not the sum, and cap concurrency with a Semaphore. Verified by timing.

The problem

Write async def fetch_all(urls, fetch, limit=10):

fetch(url) is an async function the test provides (each call takes 0.2 s)
return the responses in the same order as urls
run the calls concurrently, but never more than limit at once (use asyncio.Semaphore)

With 5 URLs and limit=10, the whole thing must take about 0.2 s, not 1.0 s.

Running in your browser: Python here runs on WebAssembly, which has no OS threads or processes. The standard APIs still work — threading, concurrent.futures, multiprocessing, asyncio — but they run on a deterministic simulator: threads run to completion when started, pools run tasks in order, and asyncio uses a virtual clock (await asyncio.sleep(0.2) advances time by 0.2 s instantly). Write exactly the code you would write in the interview.

Examples

Example 1

Input

timed_fetch(fetch_all, ['api1', 'api2', 'api3', 'api4', 'api5'])

Expected output

(['response from api1', 'response from api2', 'response from api3', 'response from api4', 'response from api5'], 0.2)

Example 2

Input

timed_fetch(fetch_all, ['a', 'b', 'c', 'd'], limit=2)[1]

Expected output

0.4

+ 1 hidden test on Submit.

Edge cases to ask about

Empty list
Order of results
Concurrency limit

How the tests call your code

These helpers run before your code. The test inputs above call them.

import asyncio

async def fake_fetch(url):
    await asyncio.sleep(0.2)
    return f"response from {url}"

def timed_fetch(fetch_all, urls, limit=10):
    async def main():
        loop = asyncio.get_running_loop()
        t0 = loop.time()
        out = await fetch_all(urls, fake_fetch, limit)
        return list(out), round(loop.time() - t0, 2)
    return asyncio.run(main())

Hints

0/3

How an interviewer scores this

0/9

1. Understand the problemRestate it, ask about edge cases and input size.
2. Write a brute-force solutionCorrect first, clever later.
3. Get the correct outputautoAll visible examples pass.
4. Analyse time complexityautoCommit to a Big-O in the Complexity Lab.
5. Analyse space complexityautoExtra memory, not the input.
6. Identify the bottleneckautoMeasure your code — where does it bend?
7. OptimiseUsually: a set or dict instead of a scan.
8. Explain why it worksSay it out loud to Solvi.
9. Test the edge casesautoSubmit passes every hidden test.

Python 3.13 · fetch_all

import asyncio

async def fetch_all(urls, fetch, limit=10):
    results = []
    for url in urls:
        results.append(await fetch(url))    # one at a time — too slow
    return results

⌘/Ctrl + Enter runs the examples

Your code runs in real CPython inside your browser — nothing is sent anywhere. The first run downloads the interpreter (about 6 MB, once). Your code is saved on this device as you type.

Complexity Lab

What does this cost as n grows?

Interviewers score the analysis as much as the code. Commit to an answer first — then check it, and read why.

Time complexity of the optimal solution

Space complexity (extra memory)

Pick both to reveal the answer.

From brute force to optimal

The progression an interviewer wants to hear, one step at a time.

Approach	Time	Space	Idea
Await in a loop	O(n · latency)	O(n)	Each call waits for the previous one.
asyncio.gather + Semaphore	O(⌈n/limit⌉ · latency)	O(n)	Overlapping I/O; the semaphore caps in-flight requests.
bestThreadPoolExecutor with requests	O(⌈n/workers⌉ · latency)	O(n)	The sync-library equivalent.

Walkthrough of the optimal approach (try it yourself first)

Network calls spend almost all their time waiting. asyncio.gather(*(one(u) for u in urls)) starts them all and waits for every result, returning them in input order. Total time ≈ the slowest call, not the sum.

An asyncio.Semaphore(limit) stops you opening 10,000 connections at once and getting rate-limited or banned. Pair it with aiohttp/httpx.AsyncClient (one shared session) and per-request timeouts. For a sync library like requests, the equivalent is a ThreadPoolExecutor — threads work well for I/O despite the GIL.

Complexity: O(n) time, O(n) space. Work and memory are linear in the number of URLs, but wall-clock time is ⌈n / limit⌉ × the per-request latency, because the waits overlap.

Reveal the reference solution

import asyncio

async def fetch_all(urls, fetch, limit=10):
    sem = asyncio.Semaphore(limit)

    async def one(url):
        async with sem:
            return await fetch(url)

    return await asyncio.gather(*(one(u) for u in urls))

Follow-ups interviewers ask

One request fails — return partial results.
Add a per-request timeout with asyncio.wait_for.

Frequently asked interview questions

Core interview concepts, complexities, and follow-ups scored by hiring teams.

What is the time complexity of Make Concurrent API Calls With asyncio in Python?

The optimal solution runs in O(n) time and O(n) auxiliary space. Work and memory are linear in the number of URLs, but wall-clock time is ⌈n / limit⌉ × the per-request latency, because the waits overlap.

What is the brute-force approach, and how do you optimise it?

Await in a loop: O(n · latency) time, O(n) space. Each call waits for the previous one. asyncio.gather + Semaphore: O(⌈n/limit⌉ · latency) time, O(n) space. Overlapping I/O; the semaphore caps in-flight requests. ThreadPoolExecutor with requests: O(⌈n/workers⌉ · latency) time, O(n) space. The sync-library equivalent.

What follow-up questions do interviewers ask about Make Concurrent API Calls With asyncio?

One request fails — return partial results. Add a per-request timeout with asyncio.wait_for.