Async By Example

calling_async_func.py

import asyncio
import time


async def square_func(n: int) -> int:
    await asyncio.sleep(2)
    print(f"square_func sleeping for 2 seconds")
    return n * n


async def double_square_func(n: int) -> int:
    res_sq = await square_func(n)
    await asyncio.sleep(2)
    print(f"double square_func sleeping for 2 seconds")
    return 2 * res_sq


async def main(*args) -> None:
    return await asyncio.gather(*(double_square_func(n) for n in args))


if __name__ == "__main__":
    t1 = time.time()
    res = asyncio.run(main(22, 34, 44))
    t2 = time.time()
    print(res)
    print(f"{t2-t1} secs")

download_pdf.py

import asyncio
import httpx
from pathlib import Path
import time
from typing import List
import aiofiles
import uvloop


def timeit(func):
    async def process(func, *args, **params):
        if asyncio.iscoroutinefunction(func):
            print("This function is a coroutine: {}".format(func.__name__))
            return await func(*args, **params)

        else:
            print("this is not a coroutine")
            return func(*args, **params)

    async def helper(*args, **params):
        print("{}.time".format(func.__name__))

        start = time.time()
        result = await process(func, *args, **params)

        print(">>>", time.time() - start)
        return result

    return helper


async def get_one(url: str) -> None:
    """Download and save one pdf file from the url."""

    fullpath = Path(url)
    fname = fullpath.name
    ext = fullpath.suffix
    client = httpx.AsyncClient()

    if not ext:
        raise RuntimeError("URL does not contain an extension")

    try:
        async with client.stream("GET", url) as r:
            async with aiofiles.open(fname, "wb") as handle:
                async for chunk in r.aiter_bytes():
                    await handle.write(chunk)

        msg = f"Finished downloading {fname}"
        return msg

    except httpx._exceptions.ConnectError:
        print("Connection error")

    except httpx._exceptions.NetworkError:
        print("NetworkError")

    except Exception as e:
        raise


@timeit
async def get_all(urls: List[str]) -> List[str]:
    res = await asyncio.gather(*(get_one(url) for url in urls))
    return res


if __name__ == "__main__":
    urls = (
        "http://www.irs.gov/pub/irs-pdf/f1040.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf",
    )
    uvloop.install()
    results = asyncio.run(get_all(urls))
    for result in results:
        print(result)