Introduction

You usually cannot magically turn a blocking third-party function into a truly asynchronous one just by writing async def around it. If the library call blocks the current thread, then calling it directly inside an async function still blocks the event loop.

What you can do is adapt the blocking function so it runs somewhere else, typically in a worker thread or process, while your async code awaits the result.

Why A Simple Wrapper Is Not Enough

This wrapper looks async, but it is still blocking:

1import asyncio
2import time
3
4
5def blocking_call():
6    time.sleep(2)
7    return "done"
8
9
10async def wrong_wrapper():
11    return blocking_call()
12
13
14async def main():
15    result = await wrong_wrapper()
16    print(result)
17
18asyncio.run(main())

wrong_wrapper is an async function syntactically, but it runs blocking_call() in the event loop thread. While that call is sleeping or performing I/O, other async tasks cannot progress.

Use asyncio.to_thread

For I/O-bound blocking work, asyncio.to_thread is usually the easiest adapter in modern Python:

1import asyncio
2import time
3
4
5def blocking_call(name):
6    time.sleep(2)
7    return f"finished {name}"
8
9
10async def call_async(name):
11    return await asyncio.to_thread(blocking_call, name)
12
13
14async def main():
15    results = await asyncio.gather(
16        call_async("job-1"),
17        call_async("job-2"),
18    )
19    print(results)
20
21asyncio.run(main())

This moves the blocking work to a thread so the event loop remains responsive.

Older Style: run_in_executor

If you need more control or support older code patterns, use the event loop executor API:

1import asyncio
2import time
3from concurrent.futures import ThreadPoolExecutor
4
5
6def blocking_call(x):
7    time.sleep(1)
8    return x * 2
9
10
11async def call_async(loop, executor, x):
12    return await loop.run_in_executor(executor, blocking_call, x)
13
14
15async def main():
16    loop = asyncio.get_running_loop()
17    with ThreadPoolExecutor(max_workers=4) as executor:
18        result = await call_async(loop, executor, 21)
19        print(result)
20
21asyncio.run(main())

This pattern is useful when you want a shared executor with a controlled number of worker threads.

CPU-Bound Versus I/O-Bound Work

Not every blocking library call belongs in a thread. If the function is CPU-heavy pure Python code, threads may not help much because of the GIL. In that case, a process pool can be the better adaptation layer.

The practical rule is simple:

• use threads for blocking I/O or libraries that spend time waiting on external systems
• consider processes for heavy CPU work
• prefer a native async library if one exists

If the third-party library already offers async support in a newer version, using that native API is better than wrapping the old synchronous one.

Wrapping A Function Call In Your Own API

A clean application-level wrapper often looks like this:

1import asyncio
2
3class Client:
4    def __init__(self, legacy_client):
5        self.legacy_client = legacy_client
6
7    async def fetch_user(self, user_id):
8        return await asyncio.to_thread(self.legacy_client.fetch_user, user_id)

Now the rest of your application can await client.fetch_user(...) without caring that the underlying library is synchronous.

What This Does Not Solve

This adaptation makes your application-level API awaitable, but it does not change the third-party library itself. The library is still blocking internally. You are just isolating that blocking behavior from the event loop.

That distinction matters because thread safety, connection pooling, and cancellation behavior still depend on the original library. For example, cancelling the await does not always stop the underlying blocking operation immediately.

Common Pitfalls

The biggest mistake is writing async def wrapper(): return blocking_func() and assuming the function is now async in a meaningful sense. It is not.

Another pitfall is sending CPU-bound work into a thread pool without measuring performance. That can still tie up resources while delivering little benefit.

A third issue is ignoring thread safety. Some synchronous clients are not safe to call concurrently from multiple threads, so wrapping them in to_thread may require external locking or separate client instances.

Summary

• You cannot make a blocking function truly non-blocking just by adding async def.
• The normal solution is to run the blocking call in a worker thread or process.
• 'asyncio.to_thread is a clean adapter for many I/O-bound library calls.'
• Use run_in_executor when you need explicit executor control.
• Prefer native async libraries when they are available.