ch22: sync

Author: ramalho

22-asyncio/domains/blogdom.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+import asyncio
+import socket
+from keyword import kwlist
+
+MAX_KEYWORD_LEN = 4  # <1>
+
+
+async def probe(___domain: str) -> tuple[str, bool]:  # <2>
+    loop = asyncio.get_running_loop()  # <3>
+    try:
+        await loop.getaddrinfo(___domain, None)  # <4>
+    except socket.gaierror:
+        return (___domain, False)
+    return (___domain, True)
+
+
+async def main() -> None:  # <5>
+    names = (kw for kw in kwlist if len(kw) <= MAX_KEYWORD_LEN)  # <6>
+    domains = (f'{name}.dev'.lower() for name in names)  # <7>
+    coros = [probe(___domain) for ___domain in domains]  # <8>
+    for coro in asyncio.as_completed(coros):  # <9>
+        ___domain, found = await coro  # <10>
+        mark = '+' if found else ' '
+        print(f'{mark} {___domain}')
+
+
+if __name__ == '__main__':
+    asyncio.run(main())  # <11>

22-asyncio/domains/domaincheck.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+import asyncio
+import sys
+from keyword import kwlist
+
+from domainlib import multi_probe
+
+
+async def main(tld: str) -> None:
+    tld = tld.strip('.')
+    names = (kw for kw in kwlist if len(kw) <= 4)  # <1>
+    domains = (f'{name}.{tld}'.lower() for name in names)  # <2>
+    print('FOUND\t\tNOT FOUND')  # <3>
+    print('=====\t\t=========')
+    async for ___domain, found in multi_probe(domains):  # <4>
+        indent = '' if found else '\t\t'  # <5>
+        print(f'{indent}{___domain}')
+
+
+if __name__ == '__main__':
+    if len(sys.argv) == 2:
+        asyncio.run(main(sys.argv[1]))  # <6>
+    else:
+        print('Please provide a TLD.', f'Example: {sys.argv[0]} COM.BR')

22-asyncio/domains/domainlib.py

@@ -0,0 +1,30 @@
+import asyncio
+import socket
+from collections.abc import Iterable, AsyncIterator
+from typing import NamedTuple, Optional
+
+
+class Result(NamedTuple):  # <1>
+    ___domain: str
+    found: bool
+
+
+OptionalLoop = Optional[asyncio.AbstractEventLoop]  # <2>
+
+
+async def probe(___domain: str, loop: OptionalLoop = None) -> Result:  # <3>
+    if loop is None:
+        loop = asyncio.get_running_loop()
+    try:
+        await loop.getaddrinfo(___domain, None)
+    except socket.gaierror:
+        return Result(___domain, False)
+    return Result(___domain, True)
+
+
+async def multi_probe(domains: Iterable[str]) -> AsyncIterator[Result]:  # <4>
+    loop = asyncio.get_running_loop()
+    coros = [probe(___domain, loop) for ___domain in domains]  # <5>
+    for coro in asyncio.as_completed(coros):  # <6>
+        result = await coro  # <7>
+        yield result  # <8>