How to handle "coroutine never awaited" as an exception?

There are many answers about how to ignore or fix this, but what I need is a solution that would handle these, because we have a huge project and something like this is bound to happen, but it's silent and therefore goes unnoticed for some time.

I want to have some function called, or an exception raised, when this happens (a coroutine somewhere wasn't awaited).

I've tried adding this, as suggested by ChatGPT, but it doesn't work:

# Set the warning mode to 'error'
import warnings
warnings.simplefilter('error', RuntimeWarning)

All it did is now the console says:

"Exception ignored in: <coroutine object XXX at 0x7fbca2e4eb90>

There is still no exception and no way to catch and react to this error.

The main problem is that this only happens when the coroutine itself goes out of scope - which is typically when the loop is shoot down, at the program ending (or, exceptionally at some "reset stage" you include in your architecture).

The error won't be raised just when a call to a co-routine is found in the code if there is a reference to it. Since one typically will want to get the return value back of a co-routine, forgetting the await just stores the co-routine itself in the variable that should contain the result, but it won't raise the error. It should be easy to catch on linters and similar tools - even static type checking - but at runtime it can be hard.

Since you want a "callback" whenever the exception is raised - maybe we can take a look at the warnings code, and see if the code responsible to raise an Exception there can be patched in runtime to make that call back instead. Ordinary Python exceptions have no way to be caught with a system-wide hook (unless a program wide hook when the exception was not handled at runtime at all - which is not the case, as the asyncio loop do handle the raised RuntimeWarning in this case to print the Exception ignored message.

But warnings are not normally raised - they are issued through calling the API in warnings - and that is Python code, which is modifiable at runtime.

Turns out that if warnings are left at default, and not changed to be raised as errors, Python warnings mechanism will first wrap it in a warnings.WarningMessage instance. It is possible to hook into the warnings module, and replace this class by one which would effect the desired callback, and patch it at runtime ("monkey patch").

So, you can code a class like the one bellow, and then run the lines following to implant a callback framework into the warnings machinery:

import warnings

OriginalMessage = warnings.WarningMessage

class HookableWarningMessage(warnings.WarningMessage):
     callback_registry = {}
     def __init__(self, *args, **kw):
          super().__init__(*args, **kw)
          self._check_callback()

     def _check_callback(self):
         if not isinstance(self.message, RuntimeWarning):
              return
         for arg in self.message.args:
             for key in self.callback_registry:
                  if key in arg:
                      self.callback_registry[key](self, self.message)
     def register(self, pattern, target):
           self.callback_registry[pattern] = target

warnings.WarningMessage = HookableWarningMessage

def not_awaited_coroutine_callback(self, RuntimeWarning):
    ...<handle warning here>...

warnings.WarningMessage.register(warnings.WarningMessage, "never awaited", not_awaited_coroutine_callback)