Deepika Pusala

Posted on Nov 11

Python Asyncio Concurrency Patterns: Semaphore, Lock, Queue, gather & TaskGroup Explained from Zero 🧵

#python #asyncio #fastapi #beginners

🧵 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

Concurrency	Parallelism
Managing multiple tasks together	Executing multiple tasks simultaneously
Tasks can take turns	Tasks can run at the same time
Can happen on one CPU thread	Usually involves multiple cores or workers

📌 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:

Both read 100
Both calculate 101
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

Feature	Lock 🔒	Semaphore 🚦
Purpose	Exclusive access	Limit concurrent access
Number allowed inside	1	N
Example	One person using a bathroom	Three people using elevators
Python	`asyncio.Lock()`	`asyncio.Semaphore(3)`

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:

📥 Receive the uploaded images
📬 Put them into a queue
👷 Have background workers pick images from the queue
⚙️ 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

Operation	Meaning
`put(item)`	Add an item to the queue
`get()`	Retrieve an item from the queue
`task_done()`	Tell the queue an item has been processed
`join()`	Wait until all queued items are marked complete
`qsize()`	Get the current approximate queue size

🐍 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()

Python opens the file
You do your work
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

Method	Purpose
`__aenter__()`	Asynchronous setup when entering the block
`__aexit__()`	Asynchronous cleanup when leaving the block

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

Normal generator	Async generator
`def`	`async def`
Uses `yield`	Uses `yield`
`for`	`async for`
Synchronous iteration	Asynchronous iteration
`next()`	`anext()`

🤷 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?

🏭 The producer puts 20 images into the queue
👷 Workers grab images from the queue
🚦 The semaphore lets only 3 processing steps run simultaneously
🔁 When one finishes, the next waiting worker proceeds
✅ 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())

Output:

['Profile fetched', 'Posts fetched', 'Notifications fetched']

The operations overlap in time, but results come back in the order you passed them in (A, B, C), not the order they finished.

💥 What if one task fails?

async def task_a():
    await asyncio.sleep(1)
    return "Success A"


async def task_b():
    await asyncio.sleep(1)
    raise ValueError("Task B failed")


async def task_c():
    await asyncio.sleep(2)
    return "Success C"


async def main():
    results = await asyncio.gather(task_a(), task_b(), task_c())
    print(results)

By default, gather() propagates the first exception to the caller.

⚠️ Important detail: the other awaitables are not automatically cancelled just because one raised an exception. They may keep running!

🛟 `gather(return_exceptions=True)`

async def main():
    results = await asyncio.gather(
        task_a(),
        task_b(),
        task_c(),
        return_exceptions=True,
    )

    for result in results:
        print(result)


asyncio.run(main())

Possible output:

Success A
Task B failed
Success C

More precisely, the second result is a ValueError exception object, not a plain string. Printing it shows its message.

You can check each result like this:

for result in results:
    if isinstance(result, Exception):
        print("Failed:", result)
    else:
        print("Success:", result)

🤷 Why is this useful?

Your social feed fetches a profile, posts, notifications, and suggested users. If notifications fail, you'd still want to show the profile and posts. One small failure shouldn't destroy the whole response!

⚠️ return_exceptions=True collects exceptions as results. It doesn't magically fix them. You still need to inspect and handle each one. It also doesn't mean cancellation or process-level failures are swallowed.

11. ⚡ Fail-fast vs Fail-soft Strategies

🎯 Important interview concept!

Imagine booking a holiday package ✈️🏨🚕: flight, hotel, and cab. The hotel booking fails. What should the app do?

🅰️ Fail-fast

If an important operation fails, stop the overall operation and report the failure instead of treating the result as successful.

If the hotel fails, you shouldn't pretend the entire package was booked. The operation should fail.

import asyncio


async def flight():
    await asyncio.sleep(1)
    return "Flight booked"


async def hotel():
    await asyncio.sleep(1)
    raise ValueError("Hotel booking failed")


async def cab():
    await asyncio.sleep(2)
    return "Cab booked"


async def main():
    try:
        results = await asyncio.gather(flight(), hotel(), cab())
        print(results)
    except ValueError as e:
        print("Booking failed:", e)


asyncio.run(main())

Output:

Booking failed: Hotel booking failed

✈️ Flight succeeded
🏨 Hotel failed
gather() propagated the hotel exception
The caller entered the except block

📌 The other tasks are not automatically cancelled by default. "Fail-fast" here means the caller gets the failure as soon as it's propagated, not necessarily that every other task stops.

Use fail-fast when:

💳 A payment process requires every step to succeed
🧾 Creating an account when required validation fails
🔗 A transaction needs several essential operations
🚫 Partial completion is unacceptable

🅱️ Fail-soft

Allow independent operations to continue even if some fail, and collect their individual outcomes.

On your social feed, if notifications fail, you still show the profile, posts, and suggested users. Only notifications are unavailable.

async def main():
    results = await asyncio.gather(
        flight(),
        hotel(),
        cab(),
        return_exceptions=True,
    )

    for result in results:
        if isinstance(result, Exception):
            print("Operation failed:", result)
        else:
            print("Operation successful:", result)

📊 Fail-fast vs Fail-soft

Feature	Fail-fast ⚡	Fail-soft 🛟
Behaviour	Propagates failure	Collects individual failures
One task fails	Caller receives exception	Other outcomes are collected
Other tasks	Not automatically cancelled by default	Continue and their results are collected
Use case	All operations are essential	Operations are independent
Python	Default `gather()`	`gather(return_exceptions=True)`

12. 🟨 JavaScript `Promise.all()` vs `Promise.allSettled()`

Since you may work with React + FastAPI, it helps to see how JavaScript handles the same ideas. 🤝

🅰️ `Promise.all()`

const promiseA = fetchProfile();
const promiseB = fetchPosts();
const promiseC = fetchNotifications();

const results = await Promise.all([promiseA, promiseB, promiseC]);

Waits for all promises to fulfill
Returns results in the original order
If one rejects, the returned Promise rejects

Similar to Python's default:

await asyncio.gather(task_a(), task_b(), task_c())

🅱️ `Promise.allSettled()`

When you want every operation's outcome, whether it succeeded or failed:

const results = await Promise.allSettled([promiseA, promiseB, promiseC]);

Possible result:

[
  { status: "fulfilled", value: "Profile fetched" },
  { status: "rejected", reason: "Posts failed" },
  { status: "fulfilled", value: "Notifications fetched" },
]

Instead of rejecting because one promise failed, Promise.allSettled() gives you the outcome of every promise. The Python equivalent is:

results = await asyncio.gather(
    task_a(),
    task_b(),
    task_c(),
    return_exceptions=True,
)

🧾 Side-by-side comparison

JavaScript	Style	Python equivalent
`Promise.all([...])`	Fail-fast ⚡	`asyncio.gather(...)`
`Promise.allSettled([...])`	Fail-soft 🛟	`asyncio.gather(..., return_exceptions=True)`

📝 Subtle differences: Promise.all() and Python's default gather() both propagate a failure, but neither automatically cancels the other operations. Also, allSettled() returns structured objects with status, value, or reason, while gather(return_exceptions=True) returns a list of plain values and exception objects. They're conceptually equivalent, not identical APIs.

13. 🧭 How to Choose the Right Primitive?

Requirement	Python primitive	What it does
Allow only one task at a time	`asyncio.Lock`	Exclusive access
Allow only N tasks at a time	`asyncio.Semaphore(N)`	Concurrency limit
Store jobs for workers	`asyncio.Queue`	Job coordination
Manage async resources	`async with`	Setup and cleanup
Produce async values	Async generator	Stream values
Run and await multiple tasks	`asyncio.gather()`	Concurrent execution
Collect failures instead of propagating	`gather(return_exceptions=True)`	Collect individual outcomes
Schedule a coroutine as a task	`asyncio.create_task()`	Schedule concurrent work

14. 🧵 How It All Fits in a FastAPI Social-Feed Project

A user uploads a post with five images. Your backend must:

📥 Receive the uploaded images
💾 Store the images
🖼️ Generate thumbnails
🗄️ Update post info in the database
🔔 Send notifications to followers

👤 User uploads a post
        │
        ▼
⚡ FastAPI endpoint → receives the request
        │
        ▼
🗄️ Database operation
   (Lock / DB transaction: protects shared in-memory state
    or database consistency, as appropriate)
        │
        ▼
🏭 Background processing
   📬 Queue      → stores image-processing jobs
   🚦 Semaphore  → limits concurrent image processing
   👷 Workers    → process images
        │
        ▼
✅ Post processing completed

And if you need to fetch profile, posts, and notifications together:

Use asyncio.gather()
Use return_exceptions=True if partial results are acceptable
Handle failures individually

15. 🎁 Bonus: `asyncio.TaskGroup`

Introduced in Python 3.11, TaskGroup gives you a structured way to create and manage related tasks.

import asyncio


async def fetch_profile():
    await asyncio.sleep(1)
    return "Profile fetched"


async def fetch_posts():
    await asyncio.sleep(2)
    return "Posts fetched"


async def main():
    async with asyncio.TaskGroup() as group:
        task1 = group.create_task(fetch_profile())
        task2 = group.create_task(fetch_posts())

    print(task1.result())
    print(task2.result())


asyncio.run(main())

A TaskGroup is created
Two tasks are scheduled
Both run concurrently
Exiting the async with block waits for all tasks to finish
Retrieve results with .result()

🆚 `TaskGroup` vs `gather()`

`asyncio.gather()`	`asyncio.TaskGroup`
Collects multiple results	Manages a group of related tasks
Default propagates the first exception	A failing child task normally cancels the remaining child tasks
Can collect exceptions with `return_exceptions=True`	Raises an `ExceptionGroup` for child failures
Great for concurrent calls and collecting results	Great for structured task management

📌 If a child task raises an ordinary exception, the other unfinished tasks in the group are cancelled, and the group waits for cleanup before raising an exception group. It's another form of fail-fast, with stronger task-lifecycle management.

16. ⚠️ Mistakes Beginners Should Avoid

1️⃣ Don't confuse concurrency with parallelism.
Asyncio helps tasks progress while others wait. It doesn't make CPU-heavy work run on multiple cores.

2️⃣ Don't use time.sleep() inside async functions.

❌ Wrong:

async def fetch_data():
    time.sleep(5)

✅ Correct:

async def fetch_data():
    await asyncio.sleep(5)

time.sleep() blocks the thread, while asyncio.sleep() lets the event loop run other tasks.

3️⃣ Don't forget to release resources.

async with lock:
    # Your operations
    pass

Prefer this over manually acquiring a lock and risking forgetting to release it.

4️⃣ Don't assume Queue handles failed jobs automatically.
If a worker fails, you need an explicit retry, logging, or failure-recording strategy. task_done() only updates the unfinished-task count. It doesn't mean the job succeeded.

5️⃣ Don't use asyncio.Lock to protect PostgreSQL across multiple servers.
Use database transactions, proper isolation, constraints, and atomic SQL operations.

6️⃣ Don't use gather() for everything.
Use it when tasks are independent. If Task B depends on Task A's result, await A first.

17. 🧪 Test Your Understanding!

Try these before peeking at the answers! ❤️

You have 100 images, but only 3 should be processed simultaneously. Which primitive helps?
- A) asyncio.Lock
- B) asyncio.Semaphore(3)
- C) asyncio.Queue
Two tasks update the same shared counter. What prevents them from interfering?
- A) asyncio.Queue
- B) asyncio.Lock
- C) asyncio.gather
You want to store 50 jobs and let background workers process them. What do you use?
- A) asyncio.Queue
- B) asyncio.Lock
- C) asyncio.sleep
Which keyword consumes an asynchronous generator?
- A) for
- B) yield
- C) async for
One API call fails, but you want the successful results from the others too. What do you use?
- A) Default asyncio.gather()
- B) asyncio.gather(return_exceptions=True)
- C) asyncio.Lock
Which JavaScript method is conceptually similar to gather(return_exceptions=True)?
- A) Promise.all()
- B) Promise.allSettled()
- C) Promise.race()
What happens when you exit an async with block?
- A) The program terminates
- B) The context manager performs asynchronous cleanup
- C) All tasks are cancelled

👀 Click to reveal answers

B: asyncio.Semaphore(3) limits concurrent processing
B: asyncio.Lock gives exclusive access
A: asyncio.Queue stores jobs for workers
C: async for
B: gather(return_exceptions=True) collects all outcomes
B: Promise.allSettled()
B: The context manager performs asynchronous cleanup

18. 📝 Final Revision Notes

Concept	One-line meaning
Concurrency	Managing multiple tasks during overlapping periods
Event loop	Schedules and manages asynchronous tasks
Semaphore	Allows a limited number of tasks simultaneously
Lock	Allows only one task into a protected section
Queue	Stores jobs for workers to process
Producer	Adds jobs to the queue
Consumer / Worker	Retrieves and processes jobs
Backpressure	Controls incoming work when capacity is limited
Async context manager	Manages asynchronous setup and cleanup
Async generator	Produces values asynchronously, one at a time
`asyncio.gather()`	Runs multiple awaitables concurrently and collects outcomes
`return_exceptions=True`	Collects failures as results
Fail-fast	Propagates failure instead of accepting partial success
Fail-soft	Lets independent operations succeed or fail separately
`Promise.all()`	JavaScript fail-fast-style promise aggregation
`Promise.allSettled()`	JavaScript aggregation of every promise outcome
`asyncio.TaskGroup`	Structured management of related async tasks

🏁 The 4 Big Ideas to Remember

🔐 Coordination: Lock and Semaphore control access to shared resources
📬 Communication: Queue transfers work between producers and consumers
🧹 Resource management: async context managers and generators manage async operations and streams
🛟 Failure management: gather, return_exceptions, and TaskGroup coordinate task outcomes and failures

If you understand these four ideas, you understand the foundation of asynchronous concurrency patterns in Python. 🎉

🚀 What's Next?

Try building a limited-concurrency image-processing system using Queue, Semaphore, workers, and error handling. Hands-on practice is where it all clicks! 💡

If this post helped you, drop a ❤️ or 🦄 and share it with a friend learning async Python. Happy coding! 👩‍💻✨

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

1. 🤔 What is Concurrency?

⚔️ Concurrency vs Parallelism

🌐 Why do we need it in FastAPI?

2. 🐍 What is asyncio?

🔑 async and await

🎛️ What is an Event Loop?

3. 🚦 Asyncio Semaphore

🛗 Real-life example

🐍 Python implementation

📊 How does it execute?

📍 Where do we use it?

🆚 Semaphore vs BoundedSemaphore

4. 🔒 Asyncio Lock

🚻 What is a Lock?

🐍 Python implementation

❌ What happens without the Lock?

📍 Where do we use Lock?

5. ⚖️ Semaphore vs Lock

6. 📬 Asyncio Queue

📖 What is a Queue?

🧰 Important asyncio.Queue operations

🐍 A complete example: 5 images, 2 workers

🏗️ The architecture

🚧 asyncio.Queue(maxsize) and Backpressure

📍 Where do we use Queue?

7. 🚪 Async Context Managers

📂 What is a context manager?

⚡ What is an async context manager?

🛠️ A custom async context manager

🧩 The two special methods

🌍 Real-world uses

8. 🔄 Async Generators

🧪 What is a generator?

⚡ What is an async generator?

🆚 Normal vs async generator

🤷 Why do we need them?

🆚 Async generator vs coroutine

9. 🏭 Async Task Queues with Concurrency Limits

🐍 Complete example

🔍 What happens?

10. 🚨 Error Handling in Async Code

🤝 What is asyncio.gather()?

💥 What if one task fails?

🛟 gather(return_exceptions=True)

🤷 Why is this useful?

11. ⚡ Fail-fast vs Fail-soft Strategies

🅰️ Fail-fast

🅱️ Fail-soft

📊 Fail-fast vs Fail-soft

12. 🟨 JavaScript Promise.all() vs Promise.allSettled()

🅰️ Promise.all()

🅱️ Promise.allSettled()

🧾 Side-by-side comparison

13. 🧭 How to Choose the Right Primitive?

14. 🧵 How It All Fits in a FastAPI Social-Feed Project

15. 🎁 Bonus: asyncio.TaskGroup

🆚 TaskGroup vs gather()

16. ⚠️ Mistakes Beginners Should Avoid

17. 🧪 Test Your Understanding!

18. 📝 Final Revision Notes

🏁 The 4 Big Ideas to Remember

🚀 What's Next?

2. 🐍 What is `asyncio`?

🔑 `async` and `await`

🆚 `Semaphore` vs `BoundedSemaphore`

🧰 Important `asyncio.Queue` operations

🚧 `asyncio.Queue(maxsize)` and Backpressure

🤝 What is `asyncio.gather()`?

🛟 `gather(return_exceptions=True)`

12. 🟨 JavaScript `Promise.all()` vs `Promise.allSettled()`

🅰️ `Promise.all()`

🅱️ `Promise.allSettled()`

15. 🎁 Bonus: `asyncio.TaskGroup`

🆚 `TaskGroup` vs `gather()`