PythonMastery
reference 4 min read · lesson 39 of 45 in Errors

RuntimeError: Event loop is closed

1 · The lesson

read

What this error means

Something asked the asyncio event loop to do work after the loop had already shut down. The coroutine, the connection or the callback outlived the loop that was supposed to run it.

It nearly always appears after your program has finished its real work, which is what makes it feel like a phantom: the output is correct and then a traceback arrives anyway.

When you see it

python
Traceback (most recent call last):
  File "main.py", line 14, in <module>
    asyncio.run(second())
  File ".../asyncio/runners.py", line 190, in run
    return runner.run(main)
RuntimeError: Event loop is closed

Or from a destructor, printed after everything else, with no line of yours in it:

python
Exception ignored in: <function _ProactorBasePipeTransport.__del__>
RuntimeError: Event loop is closed

Why it happens

  • Calling asyncio.run() more than once and reusing something from the first call. asyncio.run() creates a loop, runs the coroutine, then closes the loop. A client created inside the first call holds a reference to a loop that no longer exists.
  • A background task that was never awaited. The loop closes while it is still pending, and it complains when finalised.
  • A client left unclosed — aiohttp.ClientSession, an async database pool — whose __del__ tries to close a transport on a dead loop.
  • On Windows, the Proactor transport is finalised late, which produces the Exception ignored in: __del__ form even when your own code is tidy.

How to fix it

Call asyncio.run() once, at the top, and do everything inside it:

python
import asyncio

async def main():
    await first()
    await second()          # same loop, one entry point

asyncio.run(main())          # exactly once
+ setup added so this can run · defines first, second
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def first(*_a, **_kw):
    print('-> first() called')
    return _AutoMock('first()')
def second(*_a, **_kw):
    print('-> second() called')
    return _AutoMock('second()')

The rule that avoids nearly all of these: asyncio.run() is the boundary between sync and async code, and a program should cross it once.

Close clients with async with, so they are gone before the loop is:

python
import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:      # closed on the way out
        async with session.get(url) as r:
            return await r.json()

asyncio.run(main())
+ setup added so this can run · defines asyncio, url, session, r
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
url = _AutoMock('url')
session = _AutoMock('session')
r = _AutoMock('r')

Await your background tasks before returning. A task created and abandoned is the commonest cause:

python
async def main():
    tasks = [asyncio.create_task(fetch(u)) for u in urls]
    return await asyncio.gather(*tasks)      # nothing outlives the loop
+ setup added so this can run · defines urls, asyncio, fetch
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

urls = ["alpha", "beta", "gamma"]
asyncio = _AutoMock('asyncio')
def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

Keep a reference to every task you create. The loop holds only a weak reference, so a task with no other reference can be garbage collected mid-flight.

In pytest, let the plugin own the loop rather than calling asyncio.run() in a test:

python
import pytest

@pytest.mark.asyncio
async def test_fetch():
    assert await fetch("/health") == {"ok": True}
+ setup added so this can run · defines fetch
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

When you'd actually see this in real code

  • A script that runs two asyncio.run() calls in sequence and shares a client between them.
  • A test suite where one test closes the loop and the next one inherits a client bound to it.
  • A scraper whose ClientSession is created at module level, outside any loop.
  • A clean run on Linux that prints Exception ignored in: __del__ on Windows, from the same unclosed transport.

See Also