同步函数is_it_bad阻塞Async调用，如何改造为非阻塞？

The Problem #

You’ve got a synchronous function is_it_bad that’s blocking your async function get_genders_by_dict when called. Let’s start by looking at your original code:

def is_it_bad(word):
    try:
        res = next((item for item in all_names if str(word) in str(item["name"])))
    except:
        res = {'name':word, 'gender':2}
    return res

And it’s being invoked from your async function like this:

async def get_genders_by_dict(res):
    letters = re.compile('[^a-zA-Z\ ]')
    fname = unidecode(str(letters.sub('', res['full_name'])))
    # Calling the blocking function here causes the event loop to stall
    gender_data = is_it_bad(fname)
    # ... rest of your logic

Why It’s Blocking #

Async Python relies on an event loop to handle tasks concurrently. When you call a synchronous function directly from an async context, it pauses the entire loop until that function finishes executing—this is exactly the blocking behavior you’re seeing.

Solutions #

1. Wrap the Sync Function in a Thread Pool (Quick, Effective Fix)

The simplest way to make this non-blocking is to run the synchronous function in a separate thread using ThreadPoolExecutor. This lets the event loop keep processing other tasks while your lookup runs in the background.

First, add the necessary imports:

import asyncio
from concurrent.futures import ThreadPoolExecutor

Keep your original function, but fix that bare except to catch only the specific exception we care about (bare except catches all exceptions, which makes debugging harder):

def is_it_bad(word):
    try:
        res = next((item for item in all_names if str(word) in str(item["name"])))
    except StopIteration:  # Only catch the StopIteration thrown by next()
        res = {'name':word, 'gender':2}
    return res

Initialize a thread pool executor (you can reuse this across your codebase):

# Adjust max_workers based on your workload—4 is a safe starting point
executor = ThreadPoolExecutor(max_workers=4)

Wrap the sync function into an async-compatible version:

async def async_is_it_bad(word):
    loop = asyncio.get_running_loop()
    # Run the sync function in the thread pool
    return await loop.run_in_executor(executor, is_it_bad, word)

Now call it in your async function with await to avoid blocking:

async def get_genders_by_dict(res):
    letters = re.compile('[^a-zA-Z\ ]')
    fname = unidecode(str(letters.sub('', res['full_name'])))
    # Use the async wrapper with await
    gender_data = await async_is_it_bad(fname)
    # ... rest of your code

2. Refactor to Native Async (Better for I/O-Bound Work)

If all_names comes from an external source (like a database or API), you should replace the synchronous lookup with an async alternative. For example, if you’re using a database, switch to an async driver (like asyncpg for PostgreSQL) and rewrite the lookup to use async queries. This eliminates the need for threads entirely and is more efficient for I/O-bound tasks.

If all_names is just an in-memory list, though, the thread pool approach is perfect—no need to overcomplicate things.

Key Notes #

• Avoid bare except clauses: They catch unexpected exceptions (like KeyboardInterrupt) and hide bugs. Always catch only the specific exceptions you expect.
• Tune max_workers: Too many threads cause overhead, too few might not solve the blocking issue effectively. Test with values that match your workload.

内容的提问来源于stack exchange，提问作者Edgard Gomez Sennovskaya