simple Curio examples

Author: ramalho

22-asyncio/domains/blogdom.py renamed to 22-asyncio/domains/asyncio/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-asyncio/domains/domaincheck.py renamed to 22-asyncio/domains/asyncio/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-asyncio/domains/domainlib.py renamed to 22-asyncio/domains/asyncio/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-asyncio/domains/curio/blogdom.py

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