ch22: examples

22-asyncio/domains/netaddr.py

@@ -0,0 +1,25 @@
+import asyncio
+import socket
+from collections.abc import Iterable, AsyncIterator
+from typing import NamedTuple
+
+
+class Result(NamedTuple):
+    name: str
+    found: bool
+
+
+async def probe(loop: asyncio.AbstractEventLoop, name: str) -> Result:
+    try:
+        await loop.getaddrinfo(name, None)
+    except socket.gaierror:
+        return Result(name, False)
+    return Result(name, True)
+
+
+async def multi_probe(names: Iterable[str]) -> AsyncIterator[Result]:
+    loop = asyncio.get_running_loop()
+    coros = [probe(loop, name) for name in names]
+    for coro in asyncio.as_completed(coros):
+        result = await coro
+        yield result

22-asyncio/domains/probe.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+
+import asyncio
+import sys
+from keyword import kwlist
+
+from netaddr import multi_probe
+
+
+async def main(tld: str) -> None:
+    names = (f'{w}.{tld}'.lower() for w in kwlist if len(w) <= 4)
+    async for name, found in multi_probe(sorted(names)):
+        mark = '.' if found else '?\t\t'
+        print(f'{mark} {name}')
+
+
+if __name__ == '__main__':
+    if len(sys.argv) == 2:
+        asyncio.run(main(sys.argv[1]))
+    else:
+        print('Please provide a TLD.', f'Example: {sys.argv[0]} COM.BR')