renamed 22-async/

22-async/domains/asyncio/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-async/domains/asyncio/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-async/domains/asyncio/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>

22-async/domains/curio/blogdom.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+from curio import run, TaskGroup
+from curio.socket import getaddrinfo, gaierror
+from keyword import kwlist
+
+MAX_KEYWORD_LEN = 4  # <1>
+
+
+async def probe(domain: str) -> tuple[str, bool]:  # <2>
+    try:
+        await getaddrinfo(domain, None)  # <4>
+    except 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>
+    async with TaskGroup() as group:
+        for domain in domains:
+            await group.spawn(probe, domain)
+        async for task in group:  # <9>
+            domain, found = task.result  # <10>
+            mark = '+' if found else ' '
+            print(f'{mark} {domain}')
+
+
+if __name__ == '__main__':
+    run(main())  # <11>

22-async/domains/curio/domaincheck.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+import curio
+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)
+    domains = (f'{name}.{tld}'.lower() for name in names)
+    print('FOUND\t\tNOT FOUND')
+    print('=====\t\t=========')
+    async for domain, found in multi_probe(domains):
+        indent = '' if found else '\t\t'
+        print(f'{indent}{domain}')
+
+
+if __name__ == '__main__':
+    if len(sys.argv) == 2:
+        curio.run(main(sys.argv[1]))
+    else:
+        print('Please provide a TLD.', f'Example: {sys.argv[0]} COM.BR')

22-async/domains/curio/domainlib.py

@@ -0,0 +1,25 @@
+from curio import TaskGroup
+from curio.socket import getaddrinfo, gaierror
+from collections.abc import Iterable, AsyncIterator
+from typing import NamedTuple
+
+
+class Result(NamedTuple):
+    domain: str
+    found: bool
+
+
+async def probe(domain: str) -> Result:
+    try:
+        await getaddrinfo(domain, None)
+    except gaierror:
+        return Result(domain, False)
+    return Result(domain, True)
+
+
+async def multi_probe(domains: Iterable[str]) -> AsyncIterator[Result]:
+    async with TaskGroup() as group:
+        for domain in domains:
+            await group.spawn(probe, domain)
+        async for task in group:
+            yield task.result

22-async/domains/curio/requirements.txt

@@ -0,0 +1 @@
+curio==1.5