async def 함수를 호출했는데 코드가 실행되지 않거나 coroutine was never awaited 경고가 나온다면, 먼저 coroutine 객체를 만든 것과 실제 실행을 예약한 것을 구분해야 합니다. 이 글에서는 일반 스크립트의 진입점에는 asyncio.run(), 이미 async 함수 안에서는 await 또는 asyncio.create_task(), 여러 작업의 완료 관리에는 TaskGroup을 선택하는 기준을 설명합니다. 실행 중인 이벤트 루프에서 asyncio.run()을 다시 호출하면 안 되는 이유도 함께 살펴보겠습니다.
async_func()만 호출하면 왜 실행되지 않을까요?
async def로 정의한 함수는 호출 즉시 일반 함수처럼 본문을 끝까지 실행하지 않습니다. 호출 결과로 coroutine 객체가 만들어질 뿐이며, 이 객체를 await하거나 Task로 예약해야 이벤트 루프가 실행할 수 있습니다.
async def fetch_data():
print("작업 실행")
coroutine = fetch_data() # coroutine 객체만 생성
위 코드에는 coroutine을 실행할 await나 Task 예약이 없습니다. 그래서 객체가 정리될 때 RuntimeWarning: coroutine 'fetch_data' was never awaited가 나타날 수 있습니다. 이 경고는 기대한 비동기 작업이 실행되지 않았을 가능성을 알려주는 신호입니다. (출처: Python 공식 문서 – Coroutines and Tasks, Developing with asyncio)
일반 Python 스크립트의 최상위에서는 asyncio.run()
동기 코드로 시작하는 일반 스크립트라면 비동기 함수의 최상위 진입점을 하나 정하고 asyncio.run()으로 실행합니다.
import asyncio
async def fetch_data():
print("작업 실행")
async def main():
await fetch_data()
if __name__ == "__main__":
asyncio.run(main())
asyncio.run()은 coroutine을 이벤트 루프에서 실행하고 결과를 반환하며, 일반적인 최상위 진입점에서 루프의 시작과 종료를 관리합니다. 함수마다 반복 호출하기보다 프로그램의 경계에서 한 번 호출하고 그 안에서 필요한 작업을 await하는 편이 구조를 이해하기 쉽습니다. (출처: Python 공식 문서 – Runners)
이미 async 함수 안이라면 await 또는 create_task()
이미 실행 중인 coroutine 안에서는 asyncio.run()을 다시 호출하지 말고 await를 사용합니다.
async def main():
result = await fetch_data() # 완료될 때까지 기다림
동시에 진행할 작업을 먼저 예약하고 나중에 결과를 기다리려면 asyncio.create_task()를 사용합니다.
async def main():
task = asyncio.create_task(fetch_data())
# 다른 비동기 작업을 이곳에서 진행할 수 있습니다.
result = await task
create_task()는 현재 실행 중인 이벤트 루프에 coroutine을 예약합니다. 다만 Task를 만들었다고 완료가 보장되는 것은 아닙니다. 참조를 관리하고 마지막에 await task로 완료와 예외를 확인해야 합니다. (출처: Python 공식 문서 – Coroutines and Tasks)
여러 작업의 완료를 함께 관리할 때는 TaskGroup
서로 관련된 여러 작업을 만들고 모두 끝났는지 관리해야 한다면 TaskGroup을 검토할 수 있습니다.
async def main():
async with asyncio.TaskGroup() as group:
group.create_task(fetch_data("one"))
group.create_task(fetch_data("two"))
TaskGroup은 컨텍스트를 벗어날 때 그룹에 속한 작업들이 완료될 때까지 기다리는 구조입니다. 작업 묶음의 생명주기를 코드 범위로 표현할 수 있습니다. (출처: Python 공식 문서 – Coroutines and Tasks)
asyncio.run() 오류가 나면 실행 환경부터 확인하세요
다음 오류는 같은 스레드에서 이미 이벤트 루프가 실행 중인데 asyncio.run()을 중첩 호출할 때 발생합니다.
RuntimeError: asyncio.run() cannot be called from a running event loop
현재 코드가 일반 스크립트인지, 이미 async 함수·실행기·호스트 환경 안인지부터 나눠 보세요.
- 일반 스크립트의 동기 진입점:
asyncio.run(main()) async def내부:await some_coroutine()- 동시에 시작할 작업:
asyncio.create_task()후 작업을await - 여러 관련 작업의 묶음:
TaskGroup
이미 이벤트 루프가 제공되는 실행 환경이라면 그 환경의 진입점에서 await할 수 있도록 함수를 구성해야 합니다. 특정 웹 프레임워크나 외부 async 라이브러리마다 진입점 규칙은 다를 수 있습니다. (출처: Python 공식 문서 – Runners)
동기 작업이 이벤트 루프를 막는 경우
비동기 함수 안에 오래 걸리는 동기 파일 처리나 blocking I/O를 직접 넣으면 이벤트 루프가 다른 Task로 진행하지 못할 수 있습니다. 이때는 해당 함수의 성격과 안전성을 확인한 뒤 asyncio.to_thread()처럼 별도 스레드에서 실행하는 방식을 검토합니다.
import asyncio
def read_report():
with open("report.txt", encoding="utf-8") as file:
return file.read()
async def main():
report = await asyncio.to_thread(read_report)
print(report)
모든 작업을 자동으로 스레드로 옮기는 것이 정답은 아닙니다. 공유 상태, 예외 처리, 취소 가능성, 실제로 blocking인지 확인한 뒤 적용해야 합니다. (출처: Python 공식 문서 – Coroutines and Tasks, Developing with asyncio)
선택 기준을 한 문장으로 정리하면
async def 호출만 해 둔 상태라면 실행되지 않은 coroutine인지 먼저 확인하고, 일반 스크립트의 경계에서는 asyncio.run(), 비동기 함수 안에서는 await, 나란히 진행할 작업은 create_task(), 관련 작업의 완료를 묶어 관리할 때는 TaskGroup을 선택하면 됩니다. 이미 실행 중인 이벤트 루프에서 asyncio.run() 오류가 났다면 중첩 실행을 없애고 현재 진입점에 맞게 await 구조로 바꾸는 것이 첫 점검입니다.
다음으로 확인할 질문
create_task()로 예약한 작업의 예외와 종료를 어떻게 확인할까요?- 동기
requests나 파일 작업을asyncio안에서 실행해도 될까요? - 여러 비동기 작업을
TaskGroup과gather중 무엇으로 묶을까요?