기술 · 실용 가이드
Jupyter나 FastAPI에서 asyncio.run()이 실패하나요? 증상이 아니라 루프 소유자를 고치세요
이 예외는 소유권 경고입니다. 같은 스레드에서 호스트가 이미 이벤트 루프를 실행합니다. 그 안에서 await 하거나 진짜 동기 진입점 하나만 루프를 소유해야 합니다.
Python은 같은 스레드에서 두 번째 이벤트 루프를 시작하기 전에 오류를 냅니다. asyncio.run()이 루프 생성, 실행, 비동기 제너레이터 정리, 실행기 종료, 루프 닫기까지 소유하기 때문입니다. Notebook 커널이나 웹 서버는 이미 이 책임을 가집니다.
안전한 수정은 호출자에 따라 다릅니다. Notebook 셀이나 비동기 라우트에서는 await 합니다. 독립 스크립트는 진입점에 asyncio.run() 하나만 둡니다. 재사용 라이브러리는 비동기 API를 제공하고 실행 중 루프에서 동기 래퍼가 호출되면 명확히 실패시킵니다.
변경 전에 루프 소유자를 확인하세요
스레드, asyncio.get_running_loop() 결과, 호출한 호스트를 기록합니다. Notebook, 비동기 웹, 수명주기 훅, CLI는 계약이 다릅니다.
직접 만들지 않은 루프에 stop(), close(), run_until_complete(), asyncio.run()을 호출하지 마세요.
Notebook과 비동기 프레임워크에서는 await 하세요
Jupyter에서 result = asyncio.run(work())을 result = await work()으로 바꿉니다. FastAPI에서는 비동기 호출자가 호스트 루프에서 await 합니다.
장기 Task는 호스트 수명주기에 연결하고 시작, 취소, 종료를 정의합니다.
독립 스크립트에는 소유자 하나만 두세요
진짜 진입점에 asyncio.run(main())을 두고 공유가 필요하면 하나의 asyncio.Runner로 순서대로 실행합니다.
도우미마다 runner를 만들지 말고 내부를 await 가능하게 유지합니다.
라이브러리의 비동기와 동기 계약을 분리하세요
주요 작업은 async def fetch_async(...)로 만들고, 동기 fetch(...)는 루프가 없을 때만 asyncio.run()을 사용합니다. 루프가 있으면 await를 안내합니다.
코루틴 생성 전에 검사하고, 이미 만든 미사용 코루틴은 명시적으로 닫습니다.
실제 스레드 경계에서만 제출하세요
다른 OS 스레드의 전용 루프에는 asyncio.run_coroutine_threadsafe(coro, loop)를 쓸 수 있지만 시간 제한, 취소, 종료 책임도 정의합니다.
루프 스레드에서 future.result()를 기다리면 교착될 수 있습니다.
루프 패치를 제한된 호환 계층으로 두세요
필요해도 범위와 버전을 제한한 테스트된 어댑터로 만들고 프로세스 전역 기본값으로 쓰지 않습니다.
값이 반환되어도 컨텍스트 오류, 보류 작업, 다음 셀 실패가 남을 수 있습니다.
테스트로 수정된 경계를 증명하세요
활동 중 루프의 비동기 API와 일반 스레드의 동기 래퍼를 시험하고 값, 예외, 취소, contextvars, 반복, asyncio.all_tasks()를 확인합니다.
부정 테스트는 활동 중 루프의 동기 래퍼가 즉시 실패하고 await를 안내하는지 확인합니다.
