RuntimeError: Event loop is closed almost always fires at the very end of a program or test, with a traceback pointing at a transport's __del__ or a connection cleanup rather than your code. It means something scheduled work on an event loop that has already been closed. The three recurring causes are reusing a loop after asyncio.run closed it, calling asyncio.run more than once, and leaving tasks or transports dangling at teardown. This guide fixes each, including the pytest-asyncio variant.
Prerequisites
- Python
3.8+(asyncio.run,asyncio.all_tasks,asyncio.get_running_loop). - For the test case:
pytest-asyncio >= 0.23(theloop_scopeparameter andasyncio_default_fixture_loop_scopesetting were added in 0.23; earlier versions used theevent_loopfixture).
Solution
The state machine below shows where the error is raised — work entering a loop that has already transitioned to closed.
Run a single top-level coroutine and drain everything before the loop closes:
import asyncio
async def main() -> None:
task = asyncio.create_task(worker())
try:
await do_work()
finally:
# Cancel and await stragglers so nothing is pending when run() closes
# the loop. return_exceptions=True swallows the CancelledError each raises.
task.cancel()
await asyncio.gather(task, return_exceptions=True)
# ONE asyncio.run for the whole program. It creates a loop, runs main, closes it.
asyncio.run(main())
For pytest-asyncio, match loop_scope to the fixture scope so teardown runs on a live loop:
# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function" # 0.23+: stop recreating the loop per test
import pytest_asyncio
# scope and loop_scope agree -> setup and teardown share one loop.
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def client():
c = await open_client()
yield c
await c.aclose() # runs on the SAME loop, not a closed one
The error is always a lifetime mismatch, and the four common shapes differ only in which object outlived which loop.
Why this works
asyncio.run is documented to create a fresh event loop, run the coroutine to completion, and then close that loop before returning. Anything still bound to it — a database connection's transport, a background Task, a __del__ cleanup — fires its callback against a loop that no longer accepts work, raising the error. Draining tasks and closing transports inside main (or a fixture's teardown) guarantees nothing is left to schedule after close. In pytest-asyncio, the loop is owned by the loop_scope; if a session fixture's teardown runs after a function-scoped loop has already closed, you hit the same wall, which is why aligning the scopes is the fix. The scope-versus-loop relationship is dissected in how to scope pytest fixtures for async tests.
Edge cases and failure modes
- Calling
asyncio.runtwice. Each call closes its loop, so any object created in the first run is bound to a dead loop in the second. Use oneasyncio.runand structure work as nested coroutines, orasyncio.Runner(3.11+) to reuse one loop across severalruncalls. aiohttp/asyncpgcleanup on__del__. Connections that schedule cleanup in__del__raise this at interpreter shutdown. Alwaysawait session.close()/await conn.close()explicitly inside the loop.loop.run_until_completeafterloop.close. Reusing a manually managed loop after closing it is the non-asyncio.runform of the same bug. Do not close a loop you intend to reuse.ProactorEventLoopon Windows. Older Pythons raised this spuriously at shutdown on Windows even with correct code; upgrade to a current 3.x where it is fixed.- Cross-loop objects in pytest. An object built in a session fixture but used by a function-scoped loop straddles two loops. Match
loop_scope, and see the pytest-asyncio vs anyio scoping trade-offs for choosing a model. - A coroutine that was never scheduled. If the "dangling" object is a coroutine you built but never wrapped in a
Taskor awaited, you getRuntimeWarning: coroutine ... was never awaitedinstead of thisRuntimeError— a distinct symptom with its own fix covered in tracing "coroutine was never awaited" warnings. Confirm which one you actually have before applying a fix.
Finding which loop an object belongs to
When the error appears in teardown, the useful question is not "which line raised" but "which loop did this object attach to, and when did that loop close". Three probes answer it.
The first is to record loop identity at creation. Any object that holds a transport, a connection or a task can be tagged with the loop that created it, and printing that alongside the current loop at failure time turns a mystery into an equality check.
import asyncio
class TaggedClient:
def __init__(self):
# id() of the loop is enough to compare; keeping a reference would leak it.
self._loop_id = id(asyncio.get_running_loop())
async def close(self):
current = id(asyncio.get_running_loop())
assert current == self._loop_id, (
f"closing on loop {current}, created on {self._loop_id}"
)
The second is asyncio's own debug mode, enabled with PYTHONASYNCIODEBUG=1 or asyncio.run(main(), debug=True). It logs the creation traceback of any coroutine or task that is destroyed while pending, which is exactly the information the bare RuntimeError omits — the error tells you where the loop was used, debug mode tells you where the object came from.
$ PYTHONASYNCIODEBUG=1 pytest tests/test_client.py -q
Task was destroyed but it is pending!
task: <Task pending name='Task-4' coro=<Client.poll() running at client.py:88>
created at tests/test_client.py:31>
The third is a teardown guard that fails loudly at the point of the mistake rather than at the next test. A session-scoped fixture that asserts no tasks remain turns a delayed, confusing error into an immediate, attributable one:
import asyncio
import pytest
@pytest.fixture(autouse=True)
async def no_pending_tasks():
yield
pending = [t for t in asyncio.all_tasks() if not t.done()]
assert not pending, f"{len(pending)} task(s) still pending: {pending[:3]}"
Once you know which object is at fault, the fix follows from its category. A connection or pool created by a fixture must be closed by that same fixture, on the same loop — await pool.close() inside the fixture's teardown, never in a __del__ or an atexit handler, both of which run after the loop is gone. A background task must be cancelled and awaited, not merely cancelled: task.cancel() followed by await asyncio.gather(task, return_exceptions=True) gives the task a chance to run its cleanup while the loop still exists.
The one case with no clean fix is a third-party library that closes its transport in __del__. There the answer is to hold the object explicitly and close it yourself before the loop shuts down, so the finaliser has nothing left to do — an async with block or an explicit await client.aclose() in the fixture removes the finaliser's chance to run at the wrong time.
Frequently Asked Questions
Why does asyncio.run raise RuntimeError: Event loop is closed?asyncio.run creates a new loop, runs the coroutine, then closes that loop. Calling it twice and reusing anything bound to the first loop, or leaving tasks and transports alive when it closes, raises the error because work is scheduled on a loop that no longer exists.
How do I avoid the error with pytest-asyncio?
Match the fixture's loop_scope to its scope in pytest-asyncio 0.23 or newer so setup and teardown share one loop, and set asyncio_default_fixture_loop_scope so the loop is not recreated per test.
Why does it only appear at the end of the program?
It usually comes from teardown: a transport, connection, or task is still pending when the loop closes, so its cleanup callback fires against a closed loop. Cancel and await all tasks and close transports before the loop shuts down.
Why does the error only appear after the last test finishes?
Because the object holding the stale loop reference is only finalised at interpreter shutdown, long after the loop closed. Anything cleaned up in __del__ or atexit runs at that point, when no loop is running at all. Close such objects explicitly in the fixture that created them, and the finaliser has nothing left to do when shutdown arrives.
Related
- Tracing "coroutine was never awaited" warnings — the sibling symptom when a coroutine is dropped rather than scheduled on a dead loop.
- How to scope pytest fixtures for async tests — the
loop_scope/fixture-scope alignment that prevents the teardown-on-closed-loop failure. - pytest-asyncio vs anyio scoping trade-offs — choosing the loop-ownership model that avoids cross-loop objects.
- Interactive debugging with pdb and ipdb — dropping a breakpoint at teardown to catch which transport is still pending.
← Back to Debugging Async Code and Event Loops