ch21 examples

Author: ramalho

21-futures/demo_executor_map.py

@@ -0,0 +1,31 @@
+"""
+Experiment with ``ThreadPoolExecutor.map``
+"""
+# tag::EXECUTOR_MAP[]
+from time import sleep, strftime
+from concurrent import futures
+
+def display(*args):  # <1>
+    print(strftime('[%H:%M:%S]'), end=' ')
+    print(*args)
+
+def loiter(n):  # <2>
+    msg = '{}loiter({}): doing nothing for {}s...'
+    display(msg.format('\t'*n, n, n))
+    sleep(n)
+    msg = '{}loiter({}): done.'
+    display(msg.format('\t'*n, n))
+    return n * 10  # <3>
+
+def main():
+    display('Script starting.')
+    executor = futures.ThreadPoolExecutor(max_workers=3)  # <4>
+    results = executor.map(loiter, range(5))  # <5>
+    display('results:', results)  # <6>
+    display('Waiting for individual results:')
+    for i, result in enumerate(results):  # <7>
+        display('result {}: {}'.format(i, result))
+
+if __name__ == '__main__':
+    main()
+# end::EXECUTOR_MAP[]

21-futures/getflags/country_codes.txt

@@ -0,0 +1,8 @@
+AD AE AF AG AL AM AO AR AT AU AZ BA BB BD BE BF BG BH BI BJ BN BO BR BS BT
+BW BY BZ CA CD CF CG CH CI CL CM CN CO CR CU CV CY CZ DE DJ DK DM DZ EC EE
+EG ER ES ET FI FJ FM FR GA GB GD GE GH GM GN GQ GR GT GW GY HN HR HT HU ID
+IE IL IN IQ IR IS IT JM JO JP KE KG KH KI KM KN KP KR KW KZ LA LB LC LI LK
+LR LS LT LU LV LY MA MC MD ME MG MH MK ML MM MN MR MT MU MV MW MX MY MZ NA
+NE NG NI NL NO NP NR NZ OM PA PE PG PH PK PL PT PW PY QA RO RS RU RW SA SB
+SC SD SE SG SI SK SL SM SN SO SR SS ST SV SY SZ TD TG TH TJ TL TM TN TO TR
+TT TV TW TZ UA UG US UY UZ VA VC VE VN VU WS YE ZA ZM ZW

21-futures/getflags/downloaded/.gitignore

@@ -0,0 +1 @@
+*.gif

21-futures/getflags/flags.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+
+"""Download flags of top 20 countries by population
+
+Sequential version
+
+Sample runs (first with new domain, so no caching ever)::
+
+    $ ./flags.py
+    BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN
+    20 downloads in 26.21s
+    $ ./flags.py
+    BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN
+    20 downloads in 14.57s
+
+
+"""
+
+# tag::FLAGS_PY[]
+import os
+import time
+from typing import Callable
+
+import requests  # <1>
+
+POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
+            'MX PH VN ET EG DE IR TR CD FR').split()  # <2>
+
+BASE_URL = 'http://fluentpython.com/data/flags'       # <3>
+DEST_DIR = 'downloaded/'                              # <4>
+
+def save_flag(img: bytes, filename: str) -> None:     # <5>
+    path = os.path.join(DEST_DIR, filename)
+    with open(path, 'wb') as fp:
+        fp.write(img)
+
+def get_flag(cc: str) -> bytes:  # <6>
+    cc = cc.lower()
+    url = f'{BASE_URL}/{cc}/{cc}.gif'
+    resp = requests.get(url)
+    return resp.content
+
+def download_many(cc_list: list[str]) -> int:  # <7>
+    for cc in sorted(cc_list):                 # <8>
+        image = get_flag(cc)
+        print(cc, end=' ', flush=True)         # <9>
+        save_flag(image, cc.lower() + '.gif')
+
+    return len(cc_list)
+
+def main(downloader: Callable[[list[str]], int]) -> None:  # <10>
+    t0 = time.perf_counter()                               # <11>
+    count = downloader(POP20_CC)
+    elapsed = time.perf_counter() - t0
+    print(f'\n{count} downloads in {elapsed:.2f}s')
+
+if __name__ == '__main__':
+    main(download_many)     # <12>
+# end::FLAGS_PY[]

21-futures/getflags/flags.zip

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+
+"""Download flags of countries (with error handling).
+
+asyncio async/await version
+
+"""
+# BEGIN FLAGS2_ASYNCIO_TOP
+import asyncio
+from collections import Counter
+
+import aiohttp
+from aiohttp import web
+from aiohttp.http_exceptions import HttpProcessingError
+import tqdm  # type: ignore
+
+from flags2_common import main, HTTPStatus, Result, save_flag
+
+# default set low to avoid errors from remote site, such as
+# 503 - Service Temporarily Unavailable
+DEFAULT_CONCUR_REQ = 5
+MAX_CONCUR_REQ = 1000
+
+
+class FetchError(Exception):  # <1>
+    def __init__(self, country_code):
+        self.country_code = country_code
+
+
+async def get_flag(session, base_url, cc):  # <2>
+    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
+    async with session.get(url) as resp:
+        if resp.status == 200:
+            return await resp.read()
+        elif resp.status == 404:
+            raise web.HTTPNotFound()
+        else:
+            raise HttpProcessingError(
+                code=resp.status, message=resp.reason,
+                headers=resp.headers)
+
+
+async def download_one(session, cc, base_url, semaphore, verbose):  # <3>
+    try:
+        async with semaphore:  # <4>
+            image = await get_flag(session, base_url, cc)  # <5>
+    except web.HTTPNotFound:  # <6>
+        status = HTTPStatus.not_found
+        msg = 'not found'
+    except Exception as exc:
+        raise FetchError(cc) from exc  # <7>
+    else:
+        save_flag(image, cc.lower() + '.gif')  # <8>
+        status = HTTPStatus.ok
+        msg = 'OK'
+
+    if verbose and msg:
+        print(cc, msg)
+
+    return Result(status, cc)
+# END FLAGS2_ASYNCIO_TOP
+
+# BEGIN FLAGS2_ASYNCIO_DOWNLOAD_MANY
+async def downloader_coro(cc_list: list[str],
+                          base_url: str,
+                          verbose: bool,
+                          concur_req: int) -> Counter[HTTPStatus]:  # <1>
+    counter: Counter[HTTPStatus] = Counter()
+    semaphore = asyncio.Semaphore(concur_req)  # <2>
+    async with aiohttp.ClientSession() as session:  # <8>
+        to_do = [download_one(session, cc, base_url, semaphore, verbose)
+                 for cc in sorted(cc_list)]  # <3>
+
+        to_do_iter = asyncio.as_completed(to_do)  # <4>
+        if not verbose:
+            to_do_iter = tqdm.tqdm(to_do_iter, total=len(cc_list))  # <5>
+        for future in to_do_iter:  # <6>
+            try:
+                res = await future  # <7>
+            except FetchError as exc:  # <8>
+                country_code = exc.country_code  # <9>
+                try:
+                    if exc.__cause__ is None:
+                        error_msg = 'Unknown cause'
+                    else:
+                        error_msg = exc.__cause__.args[0]  # <10>
+                except IndexError:
+                    error_msg = exc.__cause__.__class__.__name__  # <11>
+                if verbose and error_msg:
+                    msg = '*** Error for {}: {}'
+                    print(msg.format(country_code, error_msg))
+                status = HTTPStatus.error
+            else:
+                status = res.status
+
+            counter[status] += 1  # <12>
+
+    return counter  # <13>
+
+
+def download_many(cc_list: list[str],
+                  base_url: str,
+                  verbose: bool,
+                  concur_req: int) -> Counter[HTTPStatus]:
+    coro = downloader_coro(cc_list, base_url, verbose, concur_req)
+    counts = asyncio.run(coro)  # <14>
+
+    return counts
+
+
+if __name__ == '__main__':
+    main(download_many, DEFAULT_CONCUR_REQ, MAX_CONCUR_REQ)
+# END FLAGS2_ASYNCIO_DOWNLOAD_MANY

21-futures/getflags/flags2_asyncio.py

@@ -0,0 +1,155 @@
+"""Utilities for second set of flag examples.
+"""
+
+import os
+import time
+import sys
+import string
+import argparse
+from collections import namedtuple, Counter
+from enum import Enum
+
+
+Result = namedtuple('Result', 'status data')
+
+HTTPStatus = Enum('HTTPStatus', 'ok not_found error')
+
+POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
+            'MX PH VN ET EG DE IR TR CD FR').split()
+
+DEFAULT_CONCUR_REQ = 1
+MAX_CONCUR_REQ = 1
+
+SERVERS = {
+    'REMOTE': 'http://fluentpython.com/data/flags',
+    'LOCAL':  'http://localhost:8000/flags',
+    'DELAY':  'http://localhost:8001/flags',
+    'ERROR':  'http://localhost:8002/flags',
+}
+DEFAULT_SERVER = 'LOCAL'
+
+DEST_DIR = 'downloaded/'
+COUNTRY_CODES_FILE = 'country_codes.txt'
+
+
+def save_flag(img: bytes, filename: str) -> None:
+    path = os.path.join(DEST_DIR, filename)
+    with open(path, 'wb') as fp:
+        fp.write(img)
+
+
+def initial_report(cc_list: list[str],
+                   actual_req: int,
+                   server_label: str) -> None:
+    if len(cc_list) <= 10:
+        cc_msg = ', '.join(cc_list)
+    else:
+        cc_msg = 'from {} to {}'.format(cc_list[0], cc_list[-1])
+    print('{} site: {}'.format(server_label, SERVERS[server_label]))
+    msg = 'Searching for {} flag{}: {}'
+    plural = 's' if len(cc_list) != 1 else ''
+    print(msg.format(len(cc_list), plural, cc_msg))
+    plural = 's' if actual_req != 1 else ''
+    msg = '{} concurrent connection{} will be used.'
+    print(msg.format(actual_req, plural))
+
+
+def final_report(cc_list: list[str],
+                 counter: Counter[HTTPStatus],
+                 start_time: float) -> None:
+    elapsed = time.time() - start_time
+    print('-' * 20)
+    msg = '{} flag{} downloaded.'
+    plural = 's' if counter[HTTPStatus.ok] != 1 else ''
+    print(msg.format(counter[HTTPStatus.ok], plural))
+    if counter[HTTPStatus.not_found]:
+        print(counter[HTTPStatus.not_found], 'not found.')
+    if counter[HTTPStatus.error]:
+        plural = 's' if counter[HTTPStatus.error] != 1 else ''
+        print('{} error{}.'.format(counter[HTTPStatus.error], plural))
+    print('Elapsed time: {:.2f}s'.format(elapsed))
+
+
+def expand_cc_args(every_cc: bool,
+                   all_cc: bool,
+                   cc_args: list[str],
+                   limit: int) -> list[str]:
+    codes: set[str] = set()
+    A_Z = string.ascii_uppercase
+    if every_cc:
+        codes.update(a+b for a in A_Z for b in A_Z)
+    elif all_cc:
+        with open(COUNTRY_CODES_FILE) as fp:
+            text = fp.read()
+        codes.update(text.split())
+    else:
+        for cc in (c.upper() for c in cc_args):
+            if len(cc) == 1 and cc in A_Z:
+                codes.update(cc+c for c in A_Z)
+            elif len(cc) == 2 and all(c in A_Z for c in cc):
+                codes.add(cc)
+            else:
+                msg = 'each CC argument must be A to Z or AA to ZZ.'
+                raise ValueError('*** Usage error: '+msg)
+    return sorted(codes)[:limit]
+
+
+def process_args(default_concur_req):
+    server_options = ', '.join(sorted(SERVERS))
+    parser = argparse.ArgumentParser(
+        description='Download flags for country codes. '
+        'Default: top 20 countries by population.')
+    parser.add_argument('cc', metavar='CC', nargs='*',
+        help='country code or 1st letter (eg. B for BA...BZ)')
+    parser.add_argument('-a', '--all', action='store_true',
+        help='get all available flags (AD to ZW)')
+    parser.add_argument('-e', '--every', action='store_true',
+        help='get flags for every possible code (AA...ZZ)')
+    parser.add_argument('-l', '--limit', metavar='N', type=int,
+        help='limit to N first codes', default=sys.maxsize)
+    parser.add_argument('-m', '--max_req', metavar='CONCURRENT', type=int,
+        default=default_concur_req,
+        help=f'maximum concurrent requests (default={default_concur_req})')
+    parser.add_argument('-s', '--server', metavar='LABEL',
+        default=DEFAULT_SERVER,
+        help=('Server to hit; one of ' +
+              f'{server_options} (default={DEFAULT_SERVER})'))
+    parser.add_argument('-v', '--verbose', action='store_true',
+        help='output detailed progress info')
+    args = parser.parse_args()
+    if args.max_req < 1:
+        print('*** Usage error: --max_req CONCURRENT must be >= 1')
+        parser.print_usage()
+        sys.exit(1)
+    if args.limit < 1:
+        print('*** Usage error: --limit N must be >= 1')
+        parser.print_usage()
+        sys.exit(1)
+    args.server = args.server.upper()
+    if args.server not in SERVERS:
+        print('*** Usage error: --server LABEL must be one of',
+              server_options)
+        parser.print_usage()
+        sys.exit(1)
+    try:
+        cc_list = expand_cc_args(args.every, args.all, args.cc, args.limit)
+    except ValueError as exc:
+        print(exc.args[0])
+        parser.print_usage()
+        sys.exit(1)
+
+    if not cc_list:
+        cc_list = sorted(POP20_CC)
+    return args, cc_list
+
+
+def main(download_many, default_concur_req, max_concur_req):
+    args, cc_list = process_args(default_concur_req)
+    actual_req = min(args.max_req, max_concur_req, len(cc_list))
+    initial_report(cc_list, actual_req, args.server)
+    base_url = SERVERS[args.server]
+    t0 = time.time()
+    counter = download_many(cc_list, base_url, args.verbose, actual_req)
+    assert sum(counter.values()) == len(cc_list), \
+        'some downloads are unaccounted for'
+    final_report(cc_list, counter, t0)