Mastering Python Futures: From Basic Submissions to Event-Driven Concurrency

When building modern Python applications—whether scraping web pages, fetching data from external APIs, or querying databases—IO-bound operations often slow down execution.

Python’s concurrent.futures module provides a high-level, elegant interface for running tasks asynchronously. In this guide, we'll break down what Futures are, why you need them, and how to use them effectively using a practical e-commerce product service.

---

What is a Future?

A Future represents an eventual result of an asynchronous operation.

When you launch an expensive, long-running task concurrently, your program doesn't pause to wait for the output. Instead, it instantly gets back a Future object—a low-cost proxy or standard "claim ticket."

• The Future acts as a placeholder for a result that hasn't been computed yet.
• It keeps track of the task's execution state (PENDING, RUNNING, CANCELLED, or FINISHED).
• Once the task finishes, the Future stores the return value or any exception thrown during execution.

---

Why are Futures Needed?

In standard synchronous Python execution, calling a function blocks your main thread until that function finishes:

Task 1 (2s) ──> Task 2 (3s) ──> Task 3 (1s) = 6 seconds total

When dealing with IO-bound operations (like waiting for network responses or reading disks), your CPU sits completely idle during those delays.

By offloading tasks into background threads or processes via Futures, your application can run multiple IO operations simultaneously:

Task 1 (2s) [████████]
Task 2 (3s) [████████████]
Task 3 (1s) [████]
-----------------------------------------
Total Time: 3 seconds (time of longest task)

When Should You Use Futures?

1. IO-Bound Workloads: Scraping multiple web pages, batch-calling microservices, querying multiple databases, or fetching images concurrently (ThreadPoolExecutor).
2. CPU-Bound Parallelism: Performing heavy mathematical operations or image processing across multiple CPU cores (ProcessPoolExecutor).
3. Decoupled Workflows: When you want to trigger background processing without locking up the user interface or API request-handling pipeline.

---

Setting Up Our Mock Service

Let's imagine an e-commerce backend where fetching products for a category involves network latency.

import time
from concurrent.futures import ThreadPoolExecutor, Future, as_completed, wait, ALL_COMPLETED, FIRST_COMPLETED

products = [
    {"id": 123, "category": "home appliance", "name": "microwave"},
    {"id": 456, "category": "electronics", "name": "mobile"},
    {"id": 22,  "category": "home decor", "name": "bandharwal"},
    {"id": 45,  "category": "electronics", "name": "led tv"},
    {"id": 444, "category": "electronics", "name": "camera"},
]

def get_products_by_category(category: str, max_results=10, timeout=2.0):
    print(f"Fetching: category={category}, max_results={max_results}")
    time.sleep(timeout)  # Simulating network latency

    # Simulate an error scenario if category is invalid
    if category == "error":
        raise ValueError("Invalid category requested!")

    prods = [p for p in products if p.get("category", "") == category.lower()]
    return prods[:max_results]

---

1. Non-Blocking Callbacks with add_done_callback

When you submit a task to an executor via tpe.submit(), it immediately returns a Future object without waiting for the work to complete.

If you don't want your main thread to block waiting for results, you can attach a callback function.

def callbk(fut: Future):
    print(f"Callback received result: {fut.result()}")

def callback_version():
    with ThreadPoolExecutor(max_workers=3) as tpe:
        f1 = tpe.submit(get_products_by_category, "electronics", 2)
        f2 = tpe.submit(get_products_by_category, "home decor", 2)
        f3 = tpe.submit(get_products_by_category, "home appliance", 2)

        # Register callbacks
        f1.add_done_callback(callbk)
        f2.add_done_callback(callbk)
        f3.add_done_callback(callbk)

        print("Main thread continues execution without blocking!")

Key Takeaway

add_done_callback lets you trigger logic automatically as soon as a task completes. The main thread continues running while tasks process in the background.

---

2. Dynamic Processing with as_completed

In real-world networks, tasks don't complete in the order they were started. If Task A takes 3.0 seconds and Task B takes 0.2 seconds, waiting for Task A first delays processing Task B.

as_completed() returns an iterator that yields futures as soon as they finish, regardless of submission order.

def as_completed_version():
    with ThreadPoolExecutor(3) as tpe:
        # Varying network delays
        f1 = tpe.submit(get_products_by_category, "electronics", 2, 0.8)
        f2 = tpe.submit(get_products_by_category, "home decor", 2, 3.0)
        f3 = tpe.submit(get_products_by_category, "home appliance", 2, 0.2)

        futures = (f1, f2, f3)
        for f in as_completed(futures):
            print(f"Got result: {f.result()}")

Execution Flow

1. f3 finishes first (0.2s) → Processed immediately.
2. f1 finishes second (0.8s) → Processed next.
3. f2 finishes last (3.0s) → Processed last.

This prevents slow requests from bottlenecking faster ones in your processing pipeline.

---

3. Fine-Grained Control with wait()

Sometimes you need to pause execution until a specific condition is met across a group of futures—such as when all tasks finish or when the very first one returns.

wait() returns a tuple of two sets: (done_futures, not_done_futures).

def wait_version():
    with ThreadPoolExecutor(3) as tpe:
        f1 = tpe.submit(get_products_by_category, "electronics", 2, 0.8)
        f2 = tpe.submit(get_products_by_category, "home decor", 2, 3.0)
        f3 = tpe.submit(get_products_by_category, "home appliance", 2, 0.2)

        futures = (f1, f2, f3)

        # Block until all futures finish
        done_futures, not_done_futures = wait(futures, return_when=ALL_COMPLETED)

        print("All futures completed!")
        for df in done_futures:
            print(f"Got result: {df.result()}")

        print(f"Status: done={len(done_futures)}, pending={len(not_done_futures)}")

Modes for return_when:

• ALL_COMPLETED: Blocks until every future finishes or is cancelled.
• FIRST_COMPLETED: Unblocks as soon as at least one task completes.
• FIRST_EXCEPTION: Unblocks as soon as any task raises an exception.

---

4. Bulk Processing with Executor.map

If you have a collection of inputs and want to apply a single function across all of them in parallel, tpe.map() offers a clean syntax similar to Python's built-in map().

def map_version():
    with ThreadPoolExecutor(3) as tpe:
        categories = ['electronics', 'home decor', 'home appliance']
        limits = [2, 2, 2]
        timeouts = [0.8, 3.0, 0.2]

        # Map submits all tasks automatically
        results = tpe.map(get_products_by_category, categories, limits, timeouts)

        for res in results:
            print(f"Result: {res}")

as_completed vs map:

• as_completed yields results in order of completion.
• tpe.map yields results in order of original submission, blocking iteration if an earlier task takes longer than a later task.

---

5. Handling Exceptions in Futures

When a task running inside a thread throws an exception, it does not crash your main thread immediately. Instead, the exception is caught and stored inside the Future object.

There are two primary ways to handle exceptions safely:

Approach A: Using try...except with .result()

Calling .result() on a Future that experienced an error re-raises that exception in the calling thread.

def handle_exceptions_with_result():
    with ThreadPoolExecutor(2) as tpe:
        f1 = tpe.submit(get_products_by_category, "electronics", 2, 0.5)
        f2 = tpe.submit(get_products_by_category, "error", 2, 0.5) # Will raise ValueError

        for f in as_completed([f1, f2]):
            try:
                res = f.result()
                print(f"Success: {res}")
            except Exception as exc:
                print(f"Task failed with exception: {exc}")

Approach B: Inspecting .exception() Directly

You can check if an error occurred without throwing an unhandled exception using the .exception() method.

def handle_exceptions_with_check():
    with ThreadPoolExecutor(2) as tpe:
        f1 = tpe.submit(get_products_by_category, "error", 2, 0.5)

        # Wait or process completion
        done, _ = wait([f1])

        for f in done:
            exc = f.exception()
            if exc:
                print(f"Logged error cleanly: {exc}")
            else:
                print(f"Result: {f.result()}")

---

Summary Matrix

Pattern	Best Used For	Execution Order
Callbacks	Event-driven code where tasks process independently.	As tasks finish
as_completed	Processing a stream of results as fast as they arrive.	Order of completion
wait	Synchronizing execution or reacting to partial completions.	Configurable (ALL_COMPLETED, etc.)
map	Applying a function across an iterable sequentially in order.	Submission order

By picking the right execution pattern and handling exceptions gracefully, you can build Python services that remain fast, robust, and responsive under heavy IO load.