Skip to content

Instantly share code, notes, and snippets.

@kgriffs
Last active November 2, 2022 01:52
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save kgriffs/d827db257815701a30c1a03a66e6d09f to your computer and use it in GitHub Desktop.
Save kgriffs/d827db257815701a30c1a03a66e6d09f to your computer and use it in GitHub Desktop.
Example showing how to trap a signal with a Python asyncio coroutine function and exit cleanly without logging an error.
import asyncio
import functools
import signal
import sys
import time
async def worker(shutdown_event):
try:
while True:
await asyncio.sleep(1)
except asyncio.CancelledError: # catch it so it does not get logged by the asyncio library
pass
print('Shutting down...')
async def do_shutdown(task):
# Here we could await join() or whatever
# Then also cancel one or more tasks
task.cancel()
async def listener():
shutdown_event = asyncio.Event()
task = asyncio.create_task(worker(shutdown_event))
asyncio.get_running_loop().add_signal_handler(
signal.SIGINT,
functools.partial(asyncio.create_task, do_shutdown(task))
)
await task
# Not really necessary in this example since asyncio.run() will return
# once listener() returns and then the script ends.
sys.exit(0)
asyncio.run(listener())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment