write: pg.609

Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_13__flags3_asyncio.py

@@ -0,0 +1,60 @@
+import asyncio
+import aiohttp
+
+
+@asyncio.coroutine
+def http_get(url):
+    res = yield from aiohttp.request('GET', url)
+    if res.status == 200:
+        ctype = res.headers.get('Content-type', '').lower()
+        if 'json' in ctype or url.endswith('json'):
+            data = yield from res.json()
+        else:
+            data = yield from res.read()
+        return data
+
+    elif res.status == 404:
+        raise web.HTTPNotFound()
+    else:
+        raise aiohttp.errors.HttpProcessiongError(
+            code=res.status, message=res.reason, headers=res.headers
+        )
+
+
+@asyncio.coroutine
+def get_country(base_url, cc):
+    url = '{}/{cc}/metadata.json'.format(base_url, cc=cc.lower())
+    metadata = yield from http_get(url)
+    return metadata['country']
+
+
+@asyncio.coroutine
+def get_flag(base_url, cc):
+    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
+    return (yield from http_get(url))
+
+
+@asyncio.coroutine
+def download_one(cc, base_url, semaphore, verbose):
+    try:
+        with (yield from semaphore):
+            image = yield from get_flag(base_url, cc)
+        with (yield from sempahore):
+            country = yield from get_country(base_url, cc)
+    except web.HTTPNotFound:
+        status = HTTPStatus.not_found
+        msg = 'not found'
+    except Exception as exc:
+        raise FetchError(cc) from exc
+    else:
+        country = country.replace(' ', '_')
+        filename = '{}-{}.gif'.format(country, cc)
+        loop = asyncio.get_event_loop()
+        loop.run_in_executor(None, save_flag, image, filename)
+        status = HTTPStatus.ok
+        msg = 'OK'
+
+    if verbose and msg:
+        print(cc, msg)
+
+    return Result(status, cc)

Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_14_15__tcp_charfinder.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+
+# BEGIN TCP_CHARFINDER_TOP
+import sys
+import asyncio
+from urllib.parse import urlparse, parse_qs
+
+from ex_A__charfinder import UnicodeNameIndex  # <1>
+
+CRLF = b'\r\n'
+PROMPT = b'?> '
+
+index = UnicodeNameIndex()  # <2>
+
+
+async def handle_queries(reader, writer):  # <3>
+    while True:  # <4>
+        writer.write(PROMPT)  # can't await!  # <5>
+        try:
+            await writer.drain()  # must await!  # <6>
+        except (ConnectionResetError, BrokenPipeError) as e:
+            break
+
+        data = await reader.readline()  # <7>
+        try:
+            decoded_data = data.decode('utf8', errors='ignore').strip()[
+                :-9
+            ]  # remove trailing HTTP/1.1
+            if decoded_data[:3] == 'GET':
+                parsed_data = urlparse(decoded_data)
+                query_items = parse_qs(parsed_data.query)
+                query = ['{}'.format(item) for item in query_items['query']][0]
+        except UnicodeDecodeError:  # <8>
+            query = '\x00'
+        client = writer.get_extra_info('peername')  # <9>
+        print('Received from {}: {!r}'.format(client, query))  # <10>
+        if query:
+            if ord(query[:1]) < 32:  # <11>
+                break
+            lines = list(index.find_description_strs(query))   # <12>
+            if lines:
+                writer.writelines(
+                    line.encode() + CRLF for line in lines
+                )   # <13>
+            writer.write(
+                index.status(query, len(lines)).encode() + CRLF
+            )   # <14>
+
+            try:
+                await writer.drain()  # <15>
+            except ConnectionResetError as e:
+                pass
+            print('Sent {} results'.format(len(lines)))  # <16>
+
+    print('Close the client socket')  # <17>
+    writer.close()  # <18>
+
+
+# END TCP_CHARFINDER_TOP
+
+# BEGIN TCP_CHARFINDER_MAIN
+def main(address='127.0.0.1', port=2323):  # <1>
+    port = int(port)
+    loop = asyncio.get_event_loop()
+    server_coro = asyncio.start_server(handle_queries, address, port)   # <2>
+    server = loop.run_until_complete(server_coro)   # <3>
+
+    host = server.sockets[0].getsockname()  # <4>
+    print('Serving on {}. Hit CTRL-C to stop.'.format(host))  # <5>
+    try:
+        loop.run_forever()  # <6>
+    except KeyboardInterrupt:  # CTRL+C pressed
+        pass
+
+    print('Server shutting down.')
+    server.close()  # <7>
+    loop.run_until_complete(server.wait_closed())  # <8>
+    loop.close()  # <9>
+
+
+if __name__ == '__main__':
+    asyncio.run(main(), debug=True)
+    # main(*sys.argv[1:])  # <10>
+# END TCP_CHARFINDER_MAIN

Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_14_15_b__tcp_charfinder.py

@@ -0,0 +1,60 @@
+# reference: https://github.com/fluentpython/example-code-2e/blob/aeee41988eb733a0e1423eec153c839c3840ba6b/18b-async-await/charfinder/tcp_charfinder.py
+
+import asyncio
+import sys
+from urllib.parse import urlparse, parse_qs
+
+from ex_A__charfinder import UnicodeNameIndex
+
+CRLF = b'\r\n'
+PROMPT = b'?>'
+
+index = UnicodeNameIndex()
+
+
+async def handle_queries(reader, writer):
+    writer.write(PROMPT)  # can't yield from!
+    await writer.drain()  # must yield from!
+    data = await reader.readline()
+
+    try:
+        decoded_data = data.decode().strip()[:-9]  # remove trailing HTTP/1.1
+        parsed_data = urlparse(decoded_data)
+        query_items = parse_qs(parsed_data.query)
+        query = ['{}'.format(item) for item in query_items['query']][0]
+    except UnicodeDecodeError:
+        query = '\x00'
+    client = writer.get_extra_info('peername')
+    print('Received from {}: {!r}'.format(client, query))
+    if query:
+        # if ord(query[:1]) < 32:
+        #     break
+        lines = list(index.find_description_strs(query))
+        if lines:
+            writer.writelines(line.encode() + CRLF for line in lines)
+        writer.write(index.status(query, len(lines)).encode() + CRLF)
+
+        await writer.drain()
+        print('Sent {} results'.format(len(lines)))
+
+    print('Close the client socket')
+    writer.close()
+
+
+async def main(address='127.0.0.1', port=2323):
+    port = int(port)
+
+    server = await asyncio.start_server(handle_queries, address, port)
+
+    host = server.sockets[0].getsockname()
+    print('Serving on {}. Hit CTRL-C to stop.'.format(host))
+
+    async with server:
+        await server.serve_forever()
+
+    print('Server shutting down.')
+    server.close()
+
+
+if __name__ == '__main__':
+    asyncio.run(main(*sys.argv[1:]))

Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_16__tcp_charfinder.ipynb

@@ -0,0 +1,96 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stderr",
+     "output_type": "stream",
+     "text": [
+      "/home/lucas/codes/python-fluent/Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_14_15__tcp_charfinder.py:55: DeprecationWarning: There is no current event loop\n",
+      "  loop = asyncio.get_event_loop()\n"
+     ]
+    },
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Serving on ('127.0.0.1', 2323). Hit CTRL-C to stop.\n",
+      "Received from ('127.0.0.1', 40964): 'cat face'\n",
+      "Sent 10 results\n",
+      "Close the client socket\n",
+      "Received from ('127.0.0.1', 40968): 'sun'\n",
+      "Sent 17 results\n",
+      "Close the client socket\n",
+      "Server shutting down.\n"
+     ]
+    },
+    {
+     "name": "stderr",
+     "output_type": "stream",
+     "text": [
+      "Traceback (most recent call last):\n",
+      "  File \"/home/lucas/.pyenv/versions/3.10.2/lib/python3.10/runpy.py\", line 196, in _run_module_as_main\n",
+      "    return _run_code(code, main_globals, None,\n",
+      "  File \"/home/lucas/.pyenv/versions/3.10.2/lib/python3.10/runpy.py\", line 86, in _run_code\n",
+      "    exec(code, run_globals)\n",
+      "  File \"/home/lucas/codes/python-fluent/Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_14_15__tcp_charfinder.py\", line 73, in <module>\n",
+      "    asyncio.run(main(), debug=True)\n",
+      "  File \"/home/lucas/.pyenv/versions/3.10.2/lib/python3.10/asyncio/runners.py\", line 37, in run\n",
+      "    raise ValueError(\"a coroutine was expected, got {!r}\".format(main))\n",
+      "ValueError: a coroutine was expected, got None\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "2"
+      ]
+     },
+     "execution_count": 4,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import os\n",
+    "\n",
+    "os.system('python3 -m ex_18_14_15__tcp_charfinder')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "interpreter": {
+   "hash": "b1cad6e4136aaf5b8d341537b41e4e06441496f7176c964558b716a0d04682ee"
+  },
+  "kernelspec": {
+   "display_name": "Python 3.10.2 64-bit ('python-fluent')",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.2"
+  },
+  "orig_nbformat": 4
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

Part_5__Control_Flow/Chap_18__Concurrency_with_Asyncio/ex_18_17_18__http_charfinder.py

@@ -0,0 +1,49 @@
+import sys
+import asyncio
+
+# ex_18_18__http_charfinder_home_function
+def home(request):
+    query = request.GET.get('query', '').strip()
+    print('Query: {!r}'.format(query))
+    if query:
+        descriptions = list(index.find_description(query))
+        res = '\n'.join(
+            ROW_TPL.format(**vars(descr)) for descr in descriptions
+        )
+        msg = index.status(query, len(descriptions))
+    else:
+        descriptions = []
+        res = ''
+        msg = 'Enter words describing characters.'
+
+    html = template.format(query=query, results=res, message=msg)
+    print('Sending {} results'.format(len(descriptions)))
+    return web.Response(content_type=CONTENT_TYPE, text=html)
+
+
+# ex_18_17__http_charfinder_main_and_init_functions
+@asyncio.coroutine
+def init(loop, address, port):
+    app = web.Application(loop=loop)
+    app.router.add_route('GET', '/', home)
+    handler = app.make_handler()
+    server = yield from loop.create_server(handler, address, port)
+
+    return server.sockets[0].getsockname()
+
+
+def main(address='127.0.0.1', port=8888):
+    port = int(port)
+    loop = asyncio.get_event_loop()
+    host = loop.run_until_complete(init(loop, address, port))
+    print('Serving on {}. Hit CTRL-C to stop.'.format(host))
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:  # CTRL+C pressed
+        pass
+    print('Server shutting down.')
+    loop.close()
+
+
+if __name__ == '__main__':
+    main(*sys.argv[1:])