How to Fix 'RuntimeError: Event loop is closed' in Python asyncio and aiofiles
Understanding the 'RuntimeError: Event loop is closed' Error
If you are working with asynchronous I/O in Python using asyncio and libraries like aiofiles, you might encounter the frustrating error: RuntimeError: Event loop is closed. This usually happens during application shutdown or when asynchronous objects are garbage collected after the event loop has already stopped running.
In this guide, we will analyze why this error occurs in asynchronous file logging setups and explore the best ways to fix and avoid it in your Python applications.
Why Does This Error Happen?
The primary reason you see RuntimeError: Event loop is closed—even when you didn't explicitly call a close() method—comes down to object lifecycle management and Garbage Collection (GC).
- Unclosed Resource Handles: Storing an open file object like
self.f = await aiofiles.open(...)as an instance variable keeps an active reference to an asynchronous file handle. - Garbage Collection Timing: When your Python script finishes executing or the
LogManagerobject loses its references, Python's garbage collector attempts to clean up the unclosed file handle. - Closed Event Loop: By the time garbage collection runs, the
asyncioevent loop has already shut down. Whenaiofilestries to clean up its underlying async resource on a closed event loop, Python raisesRuntimeError: Event loop is closed. - Windows Proactor Event Loop (Python 3.8+): On Windows, Python 3.8 made
ProactorEventLoopthe default. This event loop implementation is particularly strict during teardown and frequently raises this exception if pending handlers or underlying transports aren't cleanly terminated before loop closure.
How to Fix the Issue
Solution 1: Use Context Managers (Recommended)
Rather than holding an open file handle persistent inside the class instance, open and close the file asynchronously using async with whenever you write to it. Modern operating systems handle file opening and closing rapidly, making this approach fast, reliable, and thread-safe.
import datetime
from pathlib import Path
import aiofiles
class LogManager:
def __init__(self):
current_dir = Path(__file__).resolve().parent
now = datetime.datetime.now()
file_string = now.strftime("%Y-%m-%d %H-%M-%S") + ".log"
self.path = current_dir.parent / "logs" / file_string
async def write_log(self, level: str, text: str):
message = f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} [{level}] {text}\n"
async with aiofiles.open(self.path, mode="a") as f:
await f.write(message)
async def write_info(self, info_text: str):
await self.write_log("INFO", info_text)
async def write_error(self, error_text: str):
await self.write_log("ERROR", error_text)Solution 2: Properly Manage the Lifecycle with Async Context Managers
If you prefer keeping the file stream open continuously throughout your application's lifecycle to avoid repeated file open calls, wrap the manager in an asynchronous context manager (__aenter__ and __aexit__):
import datetime
from pathlib import Path
import aiofiles
class LogManager:
def __init__(self):
current_dir = Path(__file__).resolve().parent
now = datetime.datetime.now()
file_string = now.strftime("%Y-%m-%d %H-%M-%S") + ".log"
self.path = current_dir.parent / "logs" / file_string
self.f = None
async def __aenter__(self):
self.f = await aiofiles.open(self.path, mode="a")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.f:
await self.f.close()
async def write_info(self, info_text: str):
message = f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} [INFO] {info_text}\n"
await self.f.write(message)Usage:
async def main():
async with LogManager() as logger:
await logger.write_info("Service started successfully!")
asyncio.run(main())Solution 3: Use Standard Logging or Loguru
For most Python projects, standard synchronous logging libraries (such as Python's built-in logging module or modern libraries like loguru) execute disk I/O off the main async loop without blocking CPU execution significantly, preventing event loop errors altogether.
from loguru import logger
# Configure file output
logger.add("logs/{time:YYYY-MM-DD HH-mm-ss}.log", level="INFO")
logger.info("Login Service successfully started!")Summary
The RuntimeError: Event loop is closed error occurs when asynchronous resources attempt to clean up after the asyncio loop has stopped running. By utilizing async with context managers or delegating log handling to proper lifecycle hooks, you ensure resources close gracefully before the event loop shuts down.