🧵 Concurrency Patterns & Async Primitives in Python: Let's Learn from ZERO!

If you're building a FastAPI backend (like a social-feed app), you'll soon need to handle many requests, database calls, background jobs, and failures without making your code messy. 😵

This post starts from absolute zero. No concurrency knowledge needed! We'll slowly build up to Python's asyncio toolbox:

• ✅ Concurrency vs parallelism
• ✅ async / await and the event loop
• ✅ Semaphore and Lock
• ✅ Queue (producers, workers, backpressure)
• ✅ Async context managers and async generators
• ✅ Task queues with concurrency limits
• ✅ Error handling: gather, return_exceptions, fail-fast vs fail-soft
• ✅ JavaScript Promise.all() vs Promise.allSettled()
• ✅ asyncio.TaskGroup

Let's go! 🚀

1. 🤔 What is Concurrency?

Imagine you're a waiter in a restaurant 🍽️ with three customers:

• ☕ Customer A orders coffee
• 🍵 Customer B orders tea
• 🧃 Customer C orders juice

You take A's order to the kitchen. While the coffee is being prepared, you take B's order, then C's. You don't stand around doing nothing until the coffee is ready.

That's concurrency:

Managing multiple tasks during overlapping periods of time, making progress on one task while another is waiting.

⚔️ Concurrency vs Parallelism

📌 Important: asyncio primarily provides concurrency, not CPU parallelism.

🌐 Why do we need it in FastAPI?

Imagine 100 users open your social feed at the same time. Each one requests a home feed, a profile, a profile picture, and likes/comments. If your backend finishes one operation completely before starting the next, performance suffers.

With async code, while one request waits for the database, another request can make progress. ⚡

2. 🐍 What is asyncio?

asyncio is a built-in Python library for writing asynchronous programs. It helps us:

• 🏃 Run multiple asynchronous tasks
• ⏳ Wait for network/database operations without blocking the event loop
• 🔐 Control access to shared resources
• 🤝 Coordinate tasks
• 🚨 Handle asynchronous failures

🔑 async and await

async tells Python a function is asynchronous:

async def get_posts(): return ["Post 1", "Post 2"] 

This is called a coroutine function.

await means: "Wait for this operation's result, but let the event loop run other tasks while I'm waiting."

import asyncio async def get_posts(): print("Fetching posts...") await asyncio.sleep(2) print("Posts fetched!") return ["Post 1", "Post 2"] 

• async def defines a coroutine function
• asyncio.sleep(2) simulates an async waiting operation
• await suspends this coroutine while the event loop runs other work

⚠️ Important distinction: calling get_posts() only creates a coroutine object. It doesn't run to completion by itself. You must await it or schedule it as a task:

asyncio.run(get_posts()) 

🎛️ What is an Event Loop?

Think of the event loop as a manager 👔 that decides which task runs next.

• Task A is waiting for the database 🗄️
• Task B is waiting for an API response 🌍
• Task C is ready to execute ✅

The event loop pauses A and B, and runs C. That's how asyncio achieves concurrency!

📌 await does not stop your whole Python app. It suspends the current coroutine so the event loop can schedule other work.

3. 🚦 Asyncio Semaphore

Imagine 100 users request profile pictures, and you have an external image-processing service that can handle only 5 requests at a time.

If you send all 100 at once:

• 💥 The service may get overloaded
• ❌ Requests may fail
• 🐌 Response times increase
• 🚫 You may exceed API rate limits

We need a way to limit how many tasks run a section of code at the same time. That's a Semaphore.

💡 Semaphore controls how many tasks can enter a particular section of code simultaneously.

🛗 Real-life example

A mall has only 3 elevators 🛗 and 100 people want to use them. Only 3 can ride at a time. When one frees up, the next person gets in. That's a semaphore.

🐍 Python implementation

import asyncio semaphore = asyncio.Semaphore(3) async def process_image(image_id): async with semaphore: print(f"Processing image {image_id}") await asyncio.sleep(2) print(f"Finished image {image_id}") async def main(): tasks = [] for i in range(10): task = asyncio.create_task(process_image(i)) tasks.append(task) await asyncio.gather(*tasks) asyncio.run(main()) 

Step 1: create the semaphore

semaphore = asyncio.Semaphore(3) 

Only 3 tasks can be inside the protected section at once.

Step 2: acquire it

async with semaphore: 

• Try to acquire the semaphore
• If a slot is free, enter
• If all 3 slots are taken, wait
• When done, the slot is released automatically

Step 3: simulate work

await asyncio.sleep(2) 

Step 4: wait for everything

await asyncio.gather(*tasks) 

📊 How does it execute?

Semaphore limit = 3 Task 1 🟢 Running Task 2 🟢 Running Task 3 🟢 Running Task 4 🟡 Waiting Task 5 🟡 Waiting Task 6 🟡 Waiting Task 1 finishes → Task 4 gets the free slot Task 2 finishes → Task 5 gets the free slot ...and so on until all 10 tasks are done. 

📍 Where do we use it?

• 🖼️ Limiting simultaneous image uploads
• 🌍 Limiting calls to external APIs
• 🎞️ Processing multiple images
• 💰 Restricting concurrent requests to an expensive service

🆚 Semaphore vs BoundedSemaphore

asyncio.Semaphore(3) asyncio.BoundedSemaphore(3) 

A normal Semaphore lets its counter grow if you accidentally release too many times. A BoundedSemaphore raises an error if you release beyond its initial limit. Safer! 🛡️

4. 🔒 Asyncio Lock

Now something different. Your app has a shared variable:

likes_count = 100 

Two users like the same post at the same time:

1. Both read 100
2. Both calculate 101
3. Both save 101

But the correct answer is 102! 😱 This is called a race condition.

🏁 Race condition: multiple tasks access and modify shared data concurrently, and the final result depends on the order they execute in.

🚻 What is a Lock?

An asyncio.Lock ensures only one task at a time can enter a protected section.

Think of a bathroom with one key 🔑: Person A takes the key and enters, Person B waits outside, A returns the key, then B enters.

🐍 Python implementation

import asyncio likes_count = 100 lock = asyncio.Lock() async def add_like(): global likes_count async with lock: current = likes_count await asyncio.sleep(1) likes_count = current + 1 async def main(): tasks = [] for i in range(5): tasks.append(asyncio.create_task(add_like())) await asyncio.gather(*tasks) print("Total likes:", likes_count) asyncio.run(main()) 

Expected output:

Total likes: 105 

• lock = asyncio.Lock() creates the lock
• async with lock: acquires it. If another task holds it, the current task waits
• Leaving the block releases it automatically

❌ What happens without the Lock?

All five tasks read 100 before any of them writes back. They all calculate 100 + 1 = 101, so the final result is 101 instead of 105. 😬

📍 Where do we use Lock?

• 🔢 Updating shared in-memory counters
• 🗃️ Modifying shared cache data
• 📦 Protecting shared in-memory collections

⚠️ Important backend note: an asyncio.Lock protects tasks within the same event loop. It does not protect data across multiple processes or multiple servers. For real database likes, use database transactions, atomic updates, and constraints.

5. ⚖️ Semaphore vs Lock

• Lock: only one task at a time
• Semaphore: a specified number of tasks at a time

Both work with async with. ✅

6. 📬 Asyncio Queue

Users upload 100 images to your app. You don't necessarily want to process all 100 immediately. Instead:

1. 📥 Receive the uploaded images
2. 📬 Put them into a queue
3. 👷 Have background workers pick images from the queue
4. ⚙️ Process them one by one or in controlled batches

📖 What is a Queue?

A queue stores items temporarily so tasks can retrieve and process them. It generally follows FIFO: First In, First Out.

Queue: Image A → Image B → Image C 

A worker takes Image A first, then B, then C.

🧰 Important asyncio.Queue operations

🐍 A complete example: 5 images, 2 workers

import asyncio queue = asyncio.Queue() async def producer(): for i in range(1, 6): image = f"Image {i}" await queue.put(image) print("Added:", image) async def worker(worker_id): while True: image = await queue.get() try: print(f"Worker {worker_id} processing {image}") await asyncio.sleep(2) print(f"Worker {worker_id} finished {image}") finally: queue.task_done() async def main(): # Start 2 workers in the background workers = [ asyncio.create_task(worker(1)), asyncio.create_task(worker(2)), ] # Add jobs to the queue await producer() # Wait until every item has been processed await queue.join() # Workers loop forever, so cancel them when all work is done for w in workers: w.cancel() await asyncio.gather(*workers, return_exceptions=True) print("All images processed! 🎉") asyncio.run(main()) 

Part 1: create a queue

queue = asyncio.Queue() 

Part 2: the producer is a task that creates work. Here it adds five images to the queue.

Part 3: the worker (consumer) is a task that takes work from the queue and processes it.

image = await queue.get() 

If the queue is empty, the worker waits until something is added.

queue.task_done() 

This tells the queue: "I have finished processing this item." (Note that task_done() is a normal function, not a coroutine, so we don't await it.)

Part 4: wait for everything

await queue.join() 

This waits until every item added to the queue has received a matching task_done() call.

Part 5: stop the workers

w.cancel() 

Workers run infinite loops, so we cancel them once all work is complete. We use gather(..., return_exceptions=True) to collect the cancellation results without letting cancellation exceptions interrupt cleanup.

🏗️ The architecture

👥 Users upload images │ ▼ 🏭 Producer → adds image-processing jobs │ ▼ 📬 Async Queue: Image 1 → Image 2 → Image 3 → Image 4 → Image 5 │ ┌────┴────┐ ▼ ▼ 👷 Worker 1 👷 Worker 2 │ │ └────┬────┘ ▼ ✅ Processed images 

🚧 asyncio.Queue(maxsize) and Backpressure

queue = asyncio.Queue(maxsize=5) 

This creates a queue with a maximum capacity of 5. What if it already has 5 items and the producer does this?

await queue.put("Image 6") 

The producer waits until a worker removes an item and frees up space. This is called backpressure.

🌊 Backpressure means slowing down the producer when consumers can't keep up. It stops the queue from growing endlessly and eating all your memory.

📍 Where do we use Queue?

• 🖼️ Image-processing jobs
• 🔔 Sending notifications
• 📧 Email delivery
• 📊 Background report generation
• 🎞️ Processing uploaded media

⚠️ Important: asyncio.Queue is an in-memory queue. It works inside one Python process, but it doesn't give you persistent jobs across server restarts or sharing between processes. For production distributed job systems, use tools like Celery, Redis-backed queues, or other message brokers.

7. 🚪 Async Context Managers

You've seen this syntax:

async with semaphore: 

Why async with instead of an ordinary with? First, let's understand context managers.

📂 What is a context manager?

A context manager manages the beginning and ending of an operation.

with open("notes.txt", "r") as file: content = file.read() 

1. Python opens the file
2. You do your work
3. When you leave the with block, Python automatically closes the file

This is called automatic resource management. 🧹

⚡ What is an async context manager?

It does the same thing, but supports asynchronous setup and cleanup:

async with semaphore: await process_image() 

When entering, Python asynchronously acquires the semaphore. When leaving, it releases it, even if an exception occurs inside the block!

🛠️ A custom async context manager

import asyncio class DatabaseConnection: async def __aenter__(self): print("Connecting to database...") await asyncio.sleep(1) print("Database connected!") return self async def __aexit__(self, exc_type, exc_value, traceback): print("Closing database connection...") await asyncio.sleep(1) print("Connection closed!") 

Using it:

async def main(): async with DatabaseConnection() as db: print("Executing database query...") await asyncio.sleep(2) asyncio.run(main()) 

Output:

Connecting to database... Database connected! Executing database query... Closing database connection... Connection closed! 

🧩 The two special methods

The a in both names stands for asynchronous. __aexit__() also receives exception info if something failed inside the block. It can suppress the exception by returning True, but most cleanup methods should let exceptions propagate.

🌍 Real-world uses

• 🗄️ Database connections
• 🌐 HTTP client sessions
• 📄 File operations with async libraries
• 🔒 Acquiring and releasing locks
• 🔌 Managing network connections

with # Synchronous context manager async with # Asynchronous context manager 

8. 🔄 Async Generators

This looks confusing at first, but it's actually simple. Let's start with an ordinary generator.

🧪 What is a generator?

Suppose you have one million posts in your database. Loading all of them into memory is expensive. 💸 Instead, generate one item at a time using yield:

def generate_numbers(): yield 1 yield 2 yield 3 for number in generate_numbers(): print(number) 

Output:

1 2 3 

Unlike return (which exits the function), yield produces a value and pauses, then continues from where it stopped when the next value is requested.

⚡ What is an async generator?

A generator that produces values asynchronously. It uses:

• async def
• yield
• often await

import asyncio async def generate_posts(): for i in range(1, 4): await asyncio.sleep(1) yield f"Post {i}" 

We consume it with async for:

async def main(): async for post in generate_posts(): print(post) asyncio.run(main()) 

Output:

Post 1 Post 2 Post 3 

🆚 Normal vs async generator

🤷 Why do we need them?

Instead of fetching thousands of posts at once, an async generator produces results in smaller portions. Great for:

• 🗄️ Streaming database results
• 📖 Reading data asynchronously
• 📡 Streaming API responses
• 📦 Processing large datasets
• 💬 Streaming messages from a network connection

🆚 Async generator vs coroutine

async def get_post(): return "Post 1" 

☝️ A coroutine function: returns one final result.

async def get_posts(): yield "Post 1" yield "Post 2" yield "Post 3" 

☝️ An async generator function: yields multiple values over time.

• Coroutine → one final result
• Async generator → a sequence of values over time

9. 🏭 Async Task Queues with Concurrency Limits

Let's combine everything we've learned:

• 🚦 Semaphore → limits simultaneous execution
• 📬 Queue → stores jobs
• 🏭 Producer → adds jobs
• 👷 Workers → process jobs
• 🤝 gather() → waits for tasks

Goal: 20 images should enter the queue, but at most 3 should be processed at the same time.

🐍 Complete example

import asyncio queue = asyncio.Queue() semaphore = asyncio.Semaphore(3) # max 3 images processed at once async def producer(): for i in range(1, 21): await queue.put(f"Image {i}") async def process_image(image): async with semaphore: print("Processing:", image) await asyncio.sleep(2) print("Finished:", image) async def worker(): while True: image = await queue.get() try: await process_image(image) finally: queue.task_done() async def main(): # 5 workers, but the semaphore only lets 3 process at a time workers = [asyncio.create_task(worker()) for _ in range(5)] await producer() await queue.join() for w in workers: w.cancel() await asyncio.gather(*workers, return_exceptions=True) print("All 20 images processed! 🎉") asyncio.run(main()) 

🔍 What happens?

1. 🏭 The producer puts 20 images into the queue
2. 👷 Workers grab images from the queue
3. 🚦 The semaphore lets only 3 processing steps run simultaneously
4. 🔁 When one finishes, the next waiting worker proceeds
5. ✅ All 20 images get processed

💡 Subtle point: if you had exactly 3 workers, the semaphore would be redundant, since 3 workers already limit processing to 3 images. The semaphore becomes valuable when you have more workers than the allowed limit, or when several parts of your app share the same resource limit. That's why this example uses 5 workers with a limit of 3.

10. 🚨 Error Handling in Async Code

What happens when one async task fails?

• 👤 Task A → Fetch user profile
• 📝 Task B → Fetch posts
• 🔔 Task C → Fetch notifications

What if Task B fails because the database is down? 😬 There are different strategies, but first let's understand gather().

🤝 What is asyncio.gather()?

It runs multiple awaitables concurrently and collects their results:

import asyncio async def task_a(): await asyncio.sleep(2) return "Profile fetched" async def task_b(): await asyncio.sleep(1) return "Posts fetched" async def task_c(): await asyncio.sleep(3) return "Notifications fetched" async def main(): results = await asyncio.gather(task_a(), task_b(), task_c()) print(results) asyncio.run(main()Source：https://dev.to/deepika_pusala/python-asyncio-concurrency-patterns-semaphore-lock-queue-gather-taskgroup-explained-from-zero-1na8
My favorites

Related posts

I kept missing papers in my own field until I stopped reading everything

For two years I told myself I was "keeping up" with research in my area. What I was actually doing was opening arXiv, sc

0dev.to AI

AI Chatbots for Customer Support: What Actually Works

Most customers do not care whether an answer comes from a person or a chatbot. They care about getting help quickly, wit

0dev.to AI

11 things I learned building a tool on Claude Code hooks

I usually have four or five Claude Code sessions open at once, and my problem was never the agents. It was me: a session

0dev.to AI

Squeezing an RTX 3050: fast and exact LLM inference with SonicBoom

Recently I released SonicBoom v0.0.6. SonicBoom is a PyTorch alternative, and I chose my favorite GNU Guile as its first

0dev.to AI

Understanding Infrastructure as a Service (laaS): A Case Study on Amazon Web Services (AWS)

Understanding Infrastructure as a Service (IaaS): A Case Study on Amazon Web Services (AWS) Infrastructure as a Service

0dev.to AWS

A Reproducible Video Compress Workflow for Web Delivery

A video can look perfectly reasonable on a laptop and still be a poor delivery asset. The file may exceed an upload limi

0dev.to WebDev