asyncio and asyncio.gather
Implement or sketch code for asyncio and asyncio.gather. Explain the logic, complexity, and pros/cons of this approach.
Answers use simple, clear English.
Audio N/AQuick interview answer
Logic: async def coroutines await non-blocking IO. asyncio.gather runs coroutines concurrently; return_exceptions=True prevents one failure cancelling others.
Detailed answer
Logic: async def coroutines await non-blocking IO. asyncio.gather runs coroutines concurrently; return_exceptions=True prevents one failure cancelling others. Complexity notes included in code section when present. Pros: High concurrency for IO-bound workloads on single thread. Cons: CPU-bound code blocks loop — use ProcessPoolExecutor; debugging async stack traces harder. Core: async def coroutines await non-blocking IO. asyncio.gather runs coroutines concurrently; return_exceptions=True prevents one failure cancelling others. Real-time example: FastAPI endpoint gathers user DB query + Redis cache + external HTTP in parallel. Pros: High concurrency for IO-bound workloads on single thread. Cons: CPU-bound code blocks loop — use ProcessPoolExecutor; debugging async stack traces harder. Common mistakes: Calling blocking requests.get inside async def; forgetting await on gather result. Best practices: Use httpx.AsyncClient; set timeouts; limit concurrency with Semaphore. Audience level: Tech Lead.
Full explanation
async def coroutines await non-blocking IO. asyncio.gather runs coroutines concurrently; return_exceptions=True prevents one failure cancelling others.
Real example & use case
FastAPI endpoint gathers user DB query + Redis cache + external HTTP in parallel.
Pros & cons
Pros: High concurrency for IO-bound workloads on single thread. Cons: CPU-bound code blocks loop — use ProcessPoolExecutor; debugging async stack traces harder.
Code example
import asyncio
async def fetch(n: int) -> int:
await asyncio.sleep(0.1)
return n * n
async def main():
results = await asyncio.gather(fetch(2), fetch(3), fetch(4))
print(results) # [4, 9, 16]
asyncio.run(main())Practice code · python (view only · no execution)
import asyncio
async def fetch(n: int) -> int:
await asyncio.sleep(0.1)
return n * n
async def main():
results = await asyncio.gather(fetch(2), fetch(3), fetch(4))
print(results) # [4, 9, 16]
asyncio.run(main())Common mistakes
Calling blocking requests.get inside async def; forgetting await on gather result.
Best practices
Use httpx.AsyncClient; set timeouts; limit concurrency with Semaphore.
Follow-up questions
- How would you test asyncio and asyncio.gather?
- What metrics prove asyncio and asyncio.gather is healthy in prod?
- How does asyncio and asyncio.gather change at 10× traffic?