Sửa lỗi 'RuntimeWarning: coroutine was never awaited' trong Python async/await

Tình Huống Lỗi

Bạn thêm async def vào một hàm, gọi nó, và Python ném ra lỗi này lúc runtime:

RuntimeWarning: coroutine 'fetch_data' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Hàm của bạn không bao giờ chạy. Không có dữ liệu, không có tác dụng phụ — không có gì cả. Lệnh gọi trả về một coroutine object và Python âm thầm bỏ qua nó.

Đây là lỗi mà hầu như ai học async Python lần đầu cũng vấp phải.

Nguyên Nhân

Gọi một hàm async def không thực thi nó. Nó trả về một coroutine object — một tính toán đang tạm dừng nằm trong bộ nhớ, chờ được lên lịch. Không có gì chạy cho đến khi có thứ gì đó await nó.

Hai tình huống gây ra cảnh báo này:

• Gọi một hàm async mà không có await bên trong một hàm async khác
• Gọi một hàm async từ code đồng bộ thông thường mà không dùng asyncio.run()

Ví dụ gây ra lỗi

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

async def main():
    result = fetch_data()   # BUG: thiếu await
    print(result)           # in ra: <coroutine object fetch_data at 0x...>

asyncio.run(main())

Thay vì "data", bạn nhận được một coroutine object in ra console. Python cũng phát ra:

RuntimeWarning: coroutine 'fetch_data' was never awaited

Cách Sửa: Thêm await

Đặt await trước mỗi lời gọi hàm async trong ngữ cảnh async. Chỉ một từ khóa là giải quyết được vấn đề:

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

async def main():
    result = await fetch_data()   # ĐÃ SỬA
    print(result)                 # in ra: data

asyncio.run(main())

Gọi Code Async Từ Code Đồng Bộ

Trong một hàm thông thường (không phải async), bạn không thể dùng await. Hãy dùng asyncio.run() thay thế — nó tạo một event loop mới, chạy coroutine của bạn đến khi hoàn thành, và trả về kết quả:

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return "data"

def main():  # hàm sync thông thường
    result = asyncio.run(fetch_data())   # đúng
    print(result)

main()

Một lưu ý quan trọng: đừng bao giờ gọi asyncio.run() bên trong một event loop đang chạy — ví dụ như bên trong một hàm async khác, một cell Jupyter notebook, hoặc một request handler của FastAPI. Bạn sẽ gặp một lỗi khác:

RuntimeError: This event loop is already running

Trong những môi trường đó, hãy dùng await trực tiếp.

Chạy Nhiều Coroutine Đồng Thời

Cần chạy đồng thời nhiều hàm async cùng lúc? Dùng asyncio.gather(). Nó chạy tất cả chúng song song và thu thập kết quả:

import asyncio

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

async def task_b():
    await asyncio.sleep(1)
    return "B done"

async def main():
    # Cả hai task chạy song song — tổng thời gian ~1 giây, không phải 2
    results = await asyncio.gather(task_a(), task_b())
    print(results)  # ['A done', 'B done']

asyncio.run(main())

Bỏ qua await và bạn lại quay về điểm xuất phát — hai coroutine object được tạo ra, không bao giờ được thực thi, cảnh báo được phát ra.

Bẫy Với Class Method

Phương thức __init__ của Python không thể là async. Điều này khiến nhiều người bất ngờ khi thiết lập các object cần khởi tạo bất đồng bộ:

class DataFetcher:
    async def load(self):
        await asyncio.sleep(0.5)
        self.data = "loaded"

# Sai — __init__ không thể async, nên không thể await ở đây
class App:
    def __init__(self):
        fetcher = DataFetcher()
        fetcher.load()  # coroutine không bao giờ được await!

# Đúng — chuyển phần setup async vào một phương thức async riêng
class App:
    async def setup(self):
        fetcher = DataFetcher()
        await fetcher.load()
        self.fetcher = fetcher

async def main():
    app = App()
    await app.setup()

asyncio.run(main())

Tìm Đúng Dòng Lỗi Với tracemalloc

Trong một codebase lớn, cảnh báo một mình không cho bạn biết ở đâu coroutine chưa được await được tạo ra. Bật tracemalloc ở đầu script và Python sẽ bao gồm tên file và số dòng chính xác:

import tracemalloc
tracemalloc.start()

import asyncio

async def fetch_data():
    return "data"

async def main():
    fetch_data()  # không được await

asyncio.run(main())

Đầu ra cảnh báo giờ hiển thị traceback phân bổ bộ nhớ — hữu ích hơn nhiều so với việc tìm kiếm qua hàng trăm dòng code.

Biến Thành Lỗi Cứng Khi Phát Triển

Cảnh báo rất dễ bị bỏ qua, đặc biệt khi bị vùi trong output của test. Nâng cấp cảnh báo này thành một crash cứng để nó thất bại một cách rõ ràng:

import warnings
warnings.filterwarnings("error", category=RuntimeWarning)

Hoặc truyền flag trực tiếp cho Python:

python -W error::RuntimeWarning your_script.py

Bây giờ bất kỳ coroutine nào chưa được await sẽ raise exception và dừng thực thi ngay lập tức. CI sẽ bắt được nó trước khi đến production.

Xác Nhận Đã Sửa Xong

• Chạy lại script. Dòng RuntimeWarning sẽ biến mất.
• Xác nhận giá trị trả về của hàm async là dữ liệu thực sự, không phải chuỗi <coroutine object ...>.
• Thêm một assertion nhanh vào test suite của bạn:

import asyncio

async def fetch_data():
    return "data"

async def test_fetch():
    result = await fetch_data()
    assert result == "data", f"Expected 'data', got {result!r}"
    print("Test passed")

asyncio.run(test_fetch())

Tham Khảo Nhanh

• async → async: result = await func()
• sync → async: result = asyncio.run(func())
• Nhiều task đồng thời: results = await asyncio.gather(func1(), func2())
• Không tìm được dòng nào? Thêm tracemalloc.start() ở đầu script
• Trong CI/tests: chạy với -W error::RuntimeWarning để biến thành crash cứng