We can configure a custom asyncio event loop exception handler via the asyncio.get_running_loop() method.
By default, unhandled exceptions in asyncio programs cause the event loop to emit a warning and are reported using a default exception handler once the event loop is closed.
Setting a custom exception handler allows our programs to intentionally handle never-retrieved (silent) exceptions in our programs, such as using tailored reporting and logging.
In this tutorial, you will discover how to configure and use a custom exception handler for the asyncio event loop.
Let’s get started.
Asyncio Custom Exception Handling
Sometimes exceptions are raised in our asyncio programs that are not handled.
Instead, these exceptions fall through the program and are handled by a custom exception handler.
The asyncio event loop provides a mechanism to configure and use a custom exception handler.
This is helpful for a number of reasons, such as:
- Centralized error handling: By setting a custom exception handler, we can centralize the handling of exceptions that occur during asynchronous tasks. This makes it easier to manage and debug errors, as all unhandled exceptions will be routed through the same handler.
- Logging and reporting: A custom exception handler allows us to log and report errors in a consistent manner. We can customize the handler to log errors to a file, send them to a monitoring service, or take other appropriate actions based on the specific requirements of the application.
- Debugging and troubleshooting: When developing and testing asynchronous code, having a custom exception handler in place can make it easier to debug and troubleshoot errors. We can use the handler to print relevant information about the exception, such as the stack trace and the context in which it occurred, helping us identify and fix issues more quickly.
Next, let’s look at how we might configure a custom exception handler.
Run loops using all CPUs, download your FREE book to learn how.
How to Set An Asyncio Custom Exception Handler
The asyncio event loop provides a way to set a custom exception handler.
This handler is called when an exception is raised in the event loop and not handled, such as a never-retrieved exception.
You can learn more about never-retrieved exceptions (silent exceptions) in the tutorial:
The custom exception handler can be set via the loop.set_exception_handler() method.
Set handler as the new event loop exception handler.
— Asyncio Event Loop
Using this method requires first getting access to the asyncio event loop object, such as via the asyncio.get_running_loop() method.
For example:
|
1 2 3 4 5 |
... # get the event loop loop = asyncio.get_running_loop() # set the exception handler loop.set_exception_handler(exception_handler) |
The loop.set_exception_handler() method takes the name of a target function as an argument.
This handler function must take two arguments, the loop object and a context dict.
For example:
|
1 2 3 |
# define an exception handler def exception_handler(loop, context): ... |
The context dict may have many different key-value pairs with the details and context for the exception raised, such as:
- message: Error message.
- exception (optional): Exception object.
- future (optional): asyncio.Future instance.
- task (optional): asyncio.Task instance.
- handle (optional): asyncio.Handle instance.
- protocol (optional): Protocol instance.
- transport (optional): Transport instance.
- socket (optional): socket.socket instance.
- asyncgen (optional): Asynchronous generator that caused the exception.
The handler may then retrieve details from the context dict and report the exception, such as on stderr or via the Python logging infrastructure.
For example:
|
1 2 3 4 5 6 |
# define an exception handler def exception_handler(loop, context): # get the exception ex = context['exception'] # log details print(Got exception {ex}') |
Next, let’s consider some common questions about using a custom exception handler.
Frequently Asked Questions
This section lists frequently asked questions about the custom exception handler for asyncio programs.
Q. When are exceptions handled?
Exceptions are handled after the asyncio program is completed, specifically as part of shutting down the asyncio event loop.
Exceptions are not handled as they are raised.
Exceptions are not handled after the termination of their source task or coroutine.
Q. Can we use the handler to recover from exceptions?
No, the handler is called as part of shutting down the event loop.
Q. Is the custom exception handler called for each never-retrieved exception?
Yes, each exception that is raised and unhandled in the program receives an independent call to the custom handler function.
Q. Can we re-issue tasks that failed with an unhandled exception?
Yes, but not via the exception handler.
Instead, consider using a done callback function to check the status of the task and then re-issue the task.
You can see an example in the tutorial:
Q. Can we report unhandled exceptions as they occur?
Yes, but not via the exception handler.
Instead, consider using a done callback function to check for the failure of a task and report the exception.
You can see an example in the tutorial:
Now that we know how to set the custom asyncio exception handler, let’s look at some worked examples.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
Example Default Exception Handler
Before we explore how to use the default asyncio exception handler, let’s look at an example of what happens if the default exception handler is not set.
In this example, we will define a custom coroutine that fails with an unhandled excretion.
|
1 2 3 4 5 6 |
# task that does work async def work(): # block for a moment await asyncio.sleep(1) # raise a problem raise Exception('Something Bad Happened') |
We will then run this coroutine as a background task, suspend it for a moment, and then report a final message.
|
1 2 3 4 5 6 7 8 9 10 |
# main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# SuperFastPython.com # example of the default exception handler import asyncio # task that does work async def work(): # block for a moment await asyncio.sleep(1) # raise a problem raise Exception('Something Bad Happened') # main coroutine async def main(): # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') # start the event loop asyncio.run(main()) |
Running the example first starts the asyncio event loop and runs the main() coroutine.
The main coroutine runs and reports a message. A new task is then created and scheduled to run the work() coroutine in the background,
The main() coroutine then suspends for two seconds.
The work() task runs and suspends for one second. It then resumes and fails with an unhandled exception.
The exception unwinds the task, causing it to stop running.
The main() coroutine results and reports a final message.
The asyncio event loop then shuts down.
A warning message is reported indicating that an exception was raised and was not handled.
Finally, the default exception handler is then called, reporting the details of the exception that was raised and not handled.
This highlights the behavior of the default exception handler in the asyncio event loop.
|
1 2 3 4 5 |
Starting Done Task exception was never retrieved future: <Task finished name='Task-2' coro=<work() done, defined at ...:6> exception=Exception('Something Bad Happened')> Exception: Something Bad Happened |
Next, let’s look at an example with a custom exception handler.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Event Loop Custom Exception Handler
We can explore an example of setting a custom exception handler for the asyncio event loop.
In this case, we can update the above example to add a handler function that reports the details of the unhandled exception and to configure the asyncio event loop to call this function for each unhandled exception.
Firstly, we can define the custom exception handler function.
Our function will first retrieve the exception object and report its details on stdout.
|
1 2 3 4 5 6 |
# define an exception handler def exception_handler(loop, context): # get the exception ex = context['exception'] # log details print(f'Got exception {ex}') |
We can then update the main() coroutine to retrieve the current event loop object, and then register the custom exception handler function.
|
1 2 3 4 5 |
... # get the event loop loop = asyncio.get_running_loop() # set the exception handler loop.set_exception_handler(exception_handler) |
The updated main() coroutine with this change is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# main coroutine async def main(): # get the event loop loop = asyncio.get_running_loop() # set the exception handler loop.set_exception_handler(exception_handler) # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') |
Tying this together, the complete example of using a custom exception handler for the asyncio event loop is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
# SuperFastPython.com # example of using a custom event loop exception handler import asyncio # define an exception handler def exception_handler(loop, context): # get the exception ex = context['exception'] # log details print(f'Got exception {ex}') # task that does work async def work(): # block for a moment await asyncio.sleep(1) # raise a problem raise Exception('Something Bad Happened') # main coroutine async def main(): # get the event loop loop = asyncio.get_running_loop() # set the exception handler loop.set_exception_handler(exception_handler) # report a message print('Starting') # run the task _ = asyncio.create_task(work()) # block for a moment await asyncio.sleep(2) # report a message print('Done') # start the event loop asyncio.run(main()) |
Running the example first starts the asyncio event loop and runs the main() coroutine.
The main coroutine runs and registers the custom exception handler for the event loop. It then reports a message.
A new task is then created and scheduled to run the work() coroutine in the background,
The main() coroutine then suspends for two seconds.
The work() task runs and suspends for one second. It then resumes and fails with an unhandled exception.
The exception unwinds the task, causing it to stop running.
The main() coroutine results and reports a final message.
The asyncio event loop then shuts down.
Finally, the exception_handler() custom exception handler function is then called. It retrieves the exception object from the context dict and reports it on stdout.
This highlights how we can use a custom exception handler for the asyncio event loop.
|
1 2 3 |
Starting Done Got exception Something Bad Happened |
Takeaways
You now know how to configure and use a custom asyncio event loop exception handler.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Alessio Soggetti on Unsplash
Do you have any questions?