flags2 await refactoring start

Author: ramalho

17-futures/countries/flags2_await.py

@@ -0,0 +1,118 @@
+"""Download flags of countries (with error handling).
+
+asyncio version
+
+Sample run::
+
+ $ python3 flags2_asyncio.py -s ERROR -e -m 200
+ ERROR site: http://localhost:8003/flags
+ Searching for 676 flags: from AA to ZZ
+ 200 concurrent connections will be used.
+ --------------------
+ 146 flags downloaded.
+ 363 not found.
+ 167 errors.
+ Elapsed time: 2.59s
+
+"""
+# BEGIN FLAGS2_ASYNCIO_TOP
+importasyncio
+importcollections
+
+importaiohttp
+fromaiohttpimportweb
+importtqdm
+
+fromflags2_commonimportmain, 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
+
+
+classFetchError(Exception): # <1>
+def__init__(self, country_code):
+self.country_code=country_code
+
+
+@asyncio.coroutine
+defget_flag(base_url, cc): # <2>
+url='{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
+resp=yieldfromaiohttp.request('GET', url)
+ifresp.status==200:
+image=yieldfromresp.read()
+returnimage
+elifresp.status==404:
+raiseweb.HTTPNotFound()
+else:
+raiseaiohttp.HttpProcessingError(
+code=resp.status, message=resp.reason,
+headers=resp.headers)
+
+
+@asyncio.coroutine
+defdownload_one(cc, base_url, semaphore, verbose): # <3>
+try:
+with (yieldfromsemaphore): # <4>
+image=yieldfromget_flag(base_url, cc) # <5>
+exceptweb.HTTPNotFound: # <6>
+status=HTTPStatus.not_found
+msg='not found'
+exceptExceptionasexc:
+raiseFetchError(cc) fromexc# <7>
+else:
+save_flag(image, cc.lower() +'.gif') # <8>
+status=HTTPStatus.ok
+msg='OK'
+
+ifverboseandmsg:
+print(cc, msg)
+
+returnResult(status, cc)
+# END FLAGS2_ASYNCIO_TOP
+
+# BEGIN FLAGS2_ASYNCIO_DOWNLOAD_MANY
+@asyncio.coroutine
+defdownloader_coro(cc_list, base_url, verbose, concur_req): # <1>
+counter=collections.Counter()
+semaphore=asyncio.Semaphore(concur_req) # <2>
+to_do= [download_one(cc, base_url, semaphore, verbose)
+forccinsorted(cc_list)] # <3>
+
+to_do_iter=asyncio.as_completed(to_do) # <4>
+ifnotverbose:
+to_do_iter=tqdm.tqdm(to_do_iter, total=len(cc_list)) # <5>
+forfutureinto_do_iter: # <6>
+try:
+res=yieldfromfuture# <7>
+exceptFetchErrorasexc: # <8>
+country_code=exc.country_code# <9>
+try:
+error_msg=exc.__cause__.args[0] # <10>
+exceptIndexError:
+error_msg=exc.__cause__.__class__.__name__# <11>
+ifverboseanderror_msg:
+msg='*** Error for{}:{}'
+print(msg.format(country_code, error_msg))
+status=HTTPStatus.error
+else:
+status=res.status
+
+counter[status] +=1# <12>
+
+returncounter# <13>
+
+
+defdownload_many(cc_list, base_url, verbose, concur_req):
+loop=asyncio.get_event_loop()
+coro=downloader_coro(cc_list, base_url, verbose, concur_req)
+counts=loop.run_until_complete(coro) # <14>
+loop.close() # <15>
+
+returncounts
+
+
+if__name__=='__main__':
+main(download_many, DEFAULT_CONCUR_REQ, MAX_CONCUR_REQ)
+# END FLAGS2_ASYNCIO_DOWNLOAD_MANY