objectif

Empêcher l’annulation d’une sous-opération tout en annulant l’appelant.

code minimal

import asyncio

async def work():
    await asyncio.sleep(0.01)
    return "ok"

async def main():
    task = asyncio.create_task(asyncio.shield(work()))
    task.cancel()  # l'annulation ne traverse pas shield
    try:
        res = await task
        print(res == "ok")  # attendu: True
    except asyncio.CancelledError:
        print(False)

asyncio.run(main())

utilisation

import asyncio

async def critical():
    await asyncio.sleep(0.005)
    return 42

async def main():
    try:
        await asyncio.wait_for(asyncio.shield(critical()), timeout=0.1)
        print(True)
    except asyncio.TimeoutError:
        print(False)

asyncio.run(main())

variante(s) utile(s)

import asyncio
print(callable(asyncio.shield))

notes

• shield empêche la propagation d’annulation vers l’opération protégée.
• utile pour des étapes critiques (flush, cleanup) qui doivent finir.