Introduction

In Python, the most reliable way to create an awaitable function is defining it with async def. That produces a coroutine object that integrates cleanly with asyncio. You can also build custom awaitable objects, but that path should be reserved for advanced library design.

Start with async def for Normal Application Code

async def is concise, readable, and works with await, asyncio.gather, and task cancellation. It should be your default for I O bound workflows.

1import asyncio
2
3
4async def fetch_user_profile(user_id: int) -> dict:
5    # Simulate network delay
6    await asyncio.sleep(0.1)
7    return {"id": user_id, "name": f"user-{user_id}"}
8
9
10async def main() -> None:
11    profiles = await asyncio.gather(
12        fetch_user_profile(1),
13        fetch_user_profile(2),
14        fetch_user_profile(3),
15    )
16    for p in profiles:
17        print(p)
18
19
20if __name__ == "__main__":
21    asyncio.run(main())

This pattern is fully awaitable and easy to test. You can replace asyncio.sleep with real network or database calls later without changing the call site.

Build a Custom Awaitable with await

A custom class can be awaited when it implements __await__ and returns an iterator. This is useful for wrappers that expose a domain specific API but still run on the event loop.

1import asyncio
2from typing import Iterator, Any
3
4
5class DelayResult:
6    def __init__(self, seconds: float, value: str) -> None:
7        self.seconds = seconds
8        self.value = value
9
10    async def _run(self) -> str:
11        await asyncio.sleep(self.seconds)
12        return self.value
13
14    def __await__(self) -> Iterator[Any]:
15        return self._run().__await__()
16
17
18async def main() -> None:
19    result = await DelayResult(0.2, "done")
20    print(result)
21
22
23if __name__ == "__main__":
24    asyncio.run(main())

__await__ should delegate to a real coroutine to keep semantics predictable. Implementing your own iterator protocol directly is possible, but it is error prone and usually unnecessary.

Wrap Blocking Functions Correctly

If you have existing synchronous code, do not call it directly inside async def when it can block for noticeable time. Use asyncio.to_thread or an executor bridge so the event loop stays responsive.

1import asyncio
2import time
3
4
5def blocking_hash(text: str) -> int:
6    time.sleep(0.3)
7    return hash(text)
8
9
10async def hash_async(text: str) -> int:
11    return await asyncio.to_thread(blocking_hash, text)
12
13
14async def main() -> None:
15    values = await asyncio.gather(
16        hash_async("alpha"),
17        hash_async("beta"),
18        hash_async("gamma"),
19    )
20    print(values)
21
22
23if __name__ == "__main__":
24    asyncio.run(main())

This pattern turns existing CPU or blocking library code into an awaitable boundary without freezing unrelated coroutines.

Handle Cancellation and Timeouts

Awaitable code should define timeout and cancellation behavior up front. Wrap long operations with asyncio.wait_for and catch asyncio.CancelledError only when cleanup is needed. This keeps cancellation cooperative and prevents background tasks from leaking after callers stop waiting.

1import asyncio
2
3async def slow_step() -> str:
4    await asyncio.sleep(2)
5    return "finished"
6
7async def run_with_timeout() -> None:
8    try:
9        result = await asyncio.wait_for(slow_step(), timeout=0.5)
10        print(result)
11    except asyncio.TimeoutError:
12        print("operation timed out")
13
14asyncio.run(run_with_timeout())

Common Pitfalls

A common mistake is calling an async function without await and assuming it already ran. The call only creates a coroutine object. The event loop executes it when awaited or scheduled as a task.

Another pitfall is mixing sync and async APIs without boundaries. A blocking function inside async def can stall all concurrent tasks, which looks like random slowdowns under load.

A third issue is implementing __await__ incorrectly and returning non iterator objects. If you need custom behavior, wrap a coroutine and delegate __await__ exactly as shown earlier.

Summary

• Use async def as the default way to create awaitable functions.
• Awaitables integrate with asyncio when they are coroutines or define __await__.
• Keep blocking code off the event loop by using asyncio.to_thread.
• Reserve custom awaitable classes for library level abstractions.
• Test async code with real event loop execution to catch scheduling bugs early.