Skip to content

Gufo Ping Example: Series of Requests

Lets send the sequence of the ICMP echo requests and get the replies.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

The code is straightforward:

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

We need asyncio.run() to run asynchronous code, so lets import the asyncio.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Import sys module to parse the CLI argument.

Warning

We use sys.argv only for demonstration purposes. Use argsparse or alternatives in real-world applications.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Ping object holds all necessary API, so import it from gufo.ping.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Asynchronous code must be executed in the asynchronous functions, or coroutines. So we define our function as async. We expect an address to ping as the addr argument.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

First we need to create Ping object. Ping constructor offers a lots of configuration variables for fine-tuning. Refer to the Ping reference for further details. Defaults are good enough for our tutorial, so we ommited them.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Ping.iter_rtt() returns the asynchronous iterator, yielding for each attempt the:

  • In case of success: Round-trip-time (RTT), as float, measured in seconds.
  • In case of failure: None.

The only mandatory parameter is IP address. Gufo Ping detects IPv4/IPv6 usage, so all we need is to pass an address. Function may accept the additional parameters for fine-tuning, Refer to the Ping.ping reference for details.

Note

Ping.iter_rtt() is the asynchronous iterator and should be used along with async for construction.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Lets print the result of each iteration.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

To run our example from command line we need to check we're in __main__ module.

series.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio
import sys

from gufo.ping import Ping


async def main(addr: str) -> None:
    ping = Ping()
    async for r in ping.iter_rtt(addr, count=5):
        print(r)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

Lets run our asyncronous main() function via asyncio.run and pass first command-line parameter as address.

Running

Lets check the success case. Run example as:

$ sudo python3 examples/series.py 127.0.0.1
0.001022267
0.001339276
0.000560895
0.000832927
0.001220127

Ok, we got a measured RTT.

Lets check the faulty cause. RFC-5737 defines the special range of addresses - 192.0.2.0/24, which can be used as unresponsible addresses. So, lets ping a 192.0.2.1:

$ sudo python3 examples/series.py 192.0.2.1
None
None
None
None
None

After second or so of awaiting we finally got a None, which means our request is timed out.