Asynchronization

asyncio

asyncio is a library to write concurrent code using the async/await syntax. asyncio is often a perfect fit for IO-bound and high-level structured network code.If your code is CPU-bound, consider using the multiprocessing library instead.

Generator, Coroutines, Event Loop

from random import randint
import time
import asyncio

def odds(start, stop):
    for odd in range(start, stop + 1, 2):
        yield odd

def randn_sync():
    """sync version"""

    # If we run this function n times,
    # then it will take 3n seconds.
    time.sleep(3)
    return randint(1, 10)

async def randn_async():
    """async version"""

    # asyncio.sleep() and randint(1, 10) will run at the same time
    await asyncio.sleep(3)
    return randint(1, 10)

async def main():

    start = time.perf_counter()
    r = await randn_async()
    elapsed = time.perf_counter() - start
    print(f"Random number generated: {r}. Took {elapsed} seconds.")

    start = time.perf_counter()
    # Generator comprehension
    r = await asyncio.gather(*(randn_async() for _ in range(10)))
    elapsed = time.perf_counter() - start
    print(f"Random number generated: {r}. Took {elapsed} seconds.")

if __name__ == '__main__':
    # Event loop
    asyncio.run(main())

Asynchronous HTTP Downloader

# areq.py
"""Asynchronously get links embedded in multiple pages' HMTL."""

import asyncio
import logging
import re
import sys
from typing import IO
import urllib.error
import urllib.parse

import aiofiles
import aiohttp
from aiohttp import ClientSession

logging.basicConfig(
    format="%(asctime)s %(levelname)s:%(name)s: %(message)s",
    level=logging.DEBUG,
    datefmt="%H:%M:%S",
    stream=sys.stderr,
)
logger = logging.getLogger("areq")
logging.getLogger("chardet.charsetprober").disabled = True

HREF_RE = re.compile(r'href="(.*?)"')

async def fetch_html(url: str, session: ClientSession, **kwargs) -> str:
    """GET request wrapper to fetch page HTML.

    kwargs are passed to `session.request()`.
    """

    resp = await session.request(method="GET", url=url, **kwargs)
    resp.raise_for_status()
    logger.info("Got response [%s] for URL: %s", resp.status, url)
    html = await resp.text()
    return html

async def parse(url: str, session: ClientSession, **kwargs) -> set:
    """Find HREFs in the HTML of `url`."""
    found = set()
    try:
        html = await fetch_html(url=url, session=session, **kwargs)
    except (
        aiohttp.ClientError,
        aiohttp.http_exceptions.HttpProcessingError,
    ) as e:
        logger.error(
            "aiohttp exception for %s [%s]: %s",
            url,
            getattr(e, "status", None),
            getattr(e, "message", None),
        )
        return found
    except Exception as e:
        logger.exception(
            "Non-aiohttp exception occured:  %s", getattr(e, "__dict__", {})
        )
        return found
    else:
        for link in HREF_RE.findall(html):
            try:
                abslink = urllib.parse.urljoin(url, link)
            except (urllib.error.URLError, ValueError):
                logger.exception("Error parsing URL: %s", link)
                pass
            else:
                found.add(abslink)
        logger.info("Found %d links for %s", len(found), url)
        return found

async def write_one(file: IO, url: str, **kwargs) -> None:
    """Write the found HREFs from `url` to `file`."""
    res = await parse(url=url, **kwargs)
    if not res:
        return None
    async with aiofiles.open(file, "a") as f:
        for p in res:
            await f.write(f"{url}\t{p}\n")
        logger.info("Wrote results for source URL: %s", url)

async def bulk_crawl_and_write(file: IO, urls: set, **kwargs) -> None:
    """Crawl & write concurrently to `file` for multiple `urls`."""
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(
                write_one(file=file, url=url, session=session, **kwargs)
            )
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    import pathlib
    import sys

    assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
    here = pathlib.Path(__file__).parent

    with open(here.joinpath("urls.txt")) as infile:
        urls = set(map(str.strip, infile))

    outpath = here.joinpath("foundurls.txt")
    with open(outpath, "w") as outfile:
        outfile.write("source_url\tparsed_url\n")

    asyncio.run(bulk_crawl_and_write(file=outpath, urls=urls))

Reference

Async IO in Python: A Complete Walkthrough – Real Pythonrealpython

Async IO in Python: A Complete Walkthrough - Real Python