Python & FastAPI•Part 3: Asynchronous Programming & Concurrency (26 - 40)
Graceful AsyncIO Shutdown:
In Simple Words:
Catch SIGTERM and SIGINT, cancel all pending tasks, and wait for them to finish with asyncio.gather(*tasks, return_exceptions=True).
python
import signal, asyncio
async def shutdown(loop, sig):
print(f"Received exit signal {sig.name}...")
tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
for t in tasks: t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
loop.stop()