J’ai le code synchrone suivant:
from subprocess import run
def ping(hostname):
result = run(["ping", "-c", "1 ", "-w100", hostname], capture_output=True)
return result.returncode == 0
def main():
hostnames = [
"google.com",
"qwant.com",
"nimp",
]
down = [hostname for hostname in hostnames if not ping(hostname)]
return down
if __name__ == "__main__":
down = main()
assert down == ["nimp"]
Je souhaite le rendre asynchrone avec asyncio, voici ma tentative:
import asyncio
from subprocess import run
async def ping(hostname):
cmd = f"ping -c1 -w100 {hostname}"
result = await asyncio.create_subprocess_shell(cmd)
return result.returncode == 0
async def main():
hostnames = [
"google.com",
"qwant.com",
"nimp",
]
results = await asyncio.gather(*(ping(h) for h in hostnames))
down = [hostname for hostname, result in zip(hostnames, results) if result]
return down
if __name__ == "__main__":
down = asyncio.run(main())
assert down == ["nimp"]
Et le résultat:
/usr/local/lib/python3.7/asyncio/unix_events.py:861: RuntimeWarning: A loop is being detached from a child watcher with pending handlers
RuntimeWarning)
Traceback (most recent call last):
File "./async_ping.py", line 28, in <module>
assert down == ["nimp"]
AssertionError
Le problème c’est que result.returncode
est None
.