applications/15-task-queue/notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "6f2bc8e9",
   "metadata": {},
   "source": [
    "# Task Queues\n",
    "\n",
    "Task (or job) queues are an architecture of dissecting and streamlining processes with flexibilities in scheduling and concurrency of job executions.\n",
    "\n",
    "Sounds familiar? In fact, from [Part 14 - Distributed Computation](../14-distributed-computation/notebook.ipynb), we have already seen both Spark and Dask examples that leverage such a mechanism. Though they were a bit more abstracted away from us thus feel more implicit.\n",
    "\n",
    "## News Headline Parsing\n",
    "\n",
    "The subject for this part is the more pronounced and explicit task queue usages. But before that, let's start with a simple use-case where we build a script to parse news headlines from some websites without a task queue yet:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "55e69848",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|█████████████████████████████████████████████████████████████████| 5/5 [00:11<00:00,  2.30s/it]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 190 ms, sys: 31 ms, total: 221 ms\n",
      "Wall time: 11.5 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "from tqdm import tqdm\n",
    "import requests\n",
    "\n",
    "\n",
    "def fetch(domain):\n",
    "    r = requests.get(f'https://{domain}')\n",
    "    return r.text\n",
    "\n",
    "sites = [\n",
    "    'bbc.com',\n",
    "    'theguardian.com',\n",
    "    'washingtonpost.com',\n",
    "    'foxnews.com',\n",
    "    'wsj.com',\n",
    "]\n",
    "data = {}\n",
    "for domain in tqdm(sites, ncols=100):\n",
    "    data[domain] = fetch(domain)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "806f5da1",
   "metadata": {},
   "source": [
    "The iterative approach is logical yet inefficient.\n",
    "\n",
    "We will proceed to leverage an HTML parser to parse the retrieved source text from various websites."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "0b48019a",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<header aria-label=\"BBC\" id=\"orb-banner\" role=\"banner\">\n",
      " <div class=\"orb-nav-pri orb-nav-pri-white orb-nav-empty\" dir=\"ltr\" id=\"orb-header\">\n",
      "  <div class=\"orb-nav-pri-container b-r b-g-p\">\n",
      "   <div class=\"orb-nav-section orb-nav-blocks\">\n",
      "    <a href=\"https://www.bbc.co.uk\">\n",
      "     Homepage\n",
      "    </a>\n",
      "   </div>\n",
      "   <section>\n",
      "    <div class=\"orb-skip-links\">\n",
      "     <h2>\n",
      "      Accessibility links\n",
      "     </h2>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from bs4 import BeautifulSoup\n",
    "\n",
    "soup = BeautifulSoup(data['bbc.com'], 'html.parser')\n",
    "print(soup.find('body').find('header').prettify()[:400])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dd789c1b",
   "metadata": {},
   "source": [
    "[Beautiful Soup](https://www.crummy.com/software/BeautifulSoup/bs4/doc/) is an HTML/XML parsing library for Python. It builds on top of the Python built-in [XML processing modules](https://docs.python.org/3/library/xml.html) with a touch of user-friendliness to navigate and extract needed information with relative ease."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "1a4e03e2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'text': 'Belarus athlete told by grandmother not to return',\n",
       "  'link': '/news/world-europe-58104195'},\n",
       " {'text': 'What is Fox host Tucker Carlson doing in Hungary?',\n",
       "  'link': 'https://www.bbc.com/news/world-europe-58104200'},\n",
       " {'text': 'Messi will not stay at Barca say club',\n",
       "  'link': 'https://www.bbc.com/sport/football/58108298'},\n",
       " {'text': \"The most influential band you've never heard of\",\n",
       "  'link': 'https://www.bbc.com/culture/article/20210804-sparks-the-greatest-band-youve-never-heard-of'},\n",
       " {'text': 'Why reverse ageism is worse than ever',\n",
       "  'link': 'https://www.bbc.com/worklife/article/20210730-the-acute-ageism-problem-hurting-young-workers'},\n",
       " {'text': 'Ethiopian rebels take Unesco world heritage town',\n",
       "  'link': '/news/world-africa-58101912'},\n",
       " {'text': 'Can you party and stay safe from Delta?',\n",
       "  'link': '/news/world-us-canada-58080853'},\n",
       " {'text': 'Chronic illness influencers accused of faking it',\n",
       "  'link': '/news/stories-58093455'},\n",
       " {'text': 'Man City sign £100m Grealish from Villa',\n",
       "  'link': '/sport/football/57818660'},\n",
       " {'text': 'Mixed day for USA on day 13 in Tokyo',\n",
       "  'link': '/sport/olympics/58096833'}]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "titles = []\n",
    "\n",
    "for tag in soup.find_all('a', {'class': True}):\n",
    "    if 'media__link' in tag['class']:\n",
    "        titles.append({\n",
    "            'text': tag.text.strip(),\n",
    "            'link': tag['href'] or tag.parent['href'],\n",
    "        })\n",
    "\n",
    "titles[:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "786d2fdc",
   "metadata": {},
   "source": [
    "Beautiful Soup can also find relevant nodes and information through custom HTML tag-centric functions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "914a9bf7",
   "metadata": {},
   "outputs": [],
   "source": [
    "def bbc(tag):\n",
    "    return tag.name == 'a' and 'media__link' in tag.get('class', []) and tag.get('href', '').startswith('/') and tag.text.strip()\n",
    "\n",
    "def guardian(tag):\n",
    "    return tag.name == 'a' and tag.get('data-link-name') == 'article' and tag.text.strip()\n",
    "\n",
    "def wp(tag):\n",
    "    return tag.name == 'span' and tag.parent.name == 'a' and tag.text.strip()\n",
    "\n",
    "def fox(tag):\n",
    "    return tag.name == 'a' and tag.parent.name == 'h2' and 'title' in tag.parent.get('class') and tag.text.strip()\n",
    "\n",
    "def wsj(tag):\n",
    "    return any(['headline' in cls for cls in tag.get('class', [])]) and tag.text.strip()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d186aee5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'text': \"Tom Homan sounds alarm on border crisis: Biden admin needs 'wake the hell up'\",\n",
       "  'link': 'https://video.foxnews.com/v/6266695144001/'},\n",
       " {'text': 'Former CDC Director: Existing vaccines are working again Delta',\n",
       "  'link': 'https://video.foxnews.com/v/6266693467001/'},\n",
       " {'text': 'Russ Vought explains alleged ties between Biden admin and pro-CRT group',\n",
       "  'link': 'https://video.foxnews.com/v/6266694247001/'},\n",
       " {'text': 'Gutfeld: Media not bothering to address unvaccinated minority communities',\n",
       "  'link': 'https://video.foxnews.com/v/6266598477001/'},\n",
       " {'text': \"Lt. Sutton: America isn't seeing a 'spike in violent crime,' it's a 'tsunami'\",\n",
       "  'link': 'https://video.foxnews.com/v/6266599724001/'},\n",
       " {'text': 'Ingraham: A moratorium on the Constitution',\n",
       "  'link': 'https://video.foxnews.com/v/6266597187001/'},\n",
       " {'text': 'Sean Hannity calls out Biden over COVID spike at border',\n",
       "  'link': 'https://video.foxnews.com/v/6266588538001/'},\n",
       " {'text': 'Stocks open in the green during high earning season',\n",
       "  'link': 'https://video.foxbusiness.com/v/6266662908001/'},\n",
       " {'text': 'Making someone pay to live on your property is now a federal crime',\n",
       "  'link': 'https://video.foxnews.com/v/6266583006001/'},\n",
       " {'text': 'Larry Kudlow: Obama got caught with his mask off and pants down',\n",
       "  'link': 'https://video.foxnews.com/v/6266595450001/'}]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "titles = []\n",
    "\n",
    "soup = BeautifulSoup(data['foxnews.com'], 'html.parser')\n",
    "for tag in soup.find_all(fox):\n",
    "    titles.append({\n",
    "        'text': tag.text.strip(),\n",
    "        'link': tag['href'] or tag.parent['href'],\n",
    "    })\n",
    "\n",
    "titles[:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af96ba95",
   "metadata": {},
   "source": [
    "We can reasonably put things together from end-to-end:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ca8c80a1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|█████████████████████████████████████████████████████████████████| 5/5 [00:12<00:00,  2.47s/it]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 854 ms, sys: 42.6 ms, total: 897 ms\n",
      "Wall time: 12.4 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "filter_rules = {\n",
    "    'bbc.com': bbc,\n",
    "    'theguardian.com': guardian,\n",
    "    'washingtonpost.com': wp,\n",
    "    'foxnews.com': fox,\n",
    "    'wsj.com': wsj,\n",
    "}\n",
    "titles = []\n",
    "\n",
    "for domain in tqdm(sites, ncols=100):\n",
    "    text = fetch(domain)\n",
    "    soup = BeautifulSoup(text, 'html.parser')\n",
    "    for tag in soup.find_all(filter_rules[domain]):\n",
    "        titles.append({\n",
    "            'text': tag.text.strip(),\n",
    "            'link': tag.get('href') or tag.parent.get('href'),\n",
    "            'domain': domain,\n",
    "        })"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ed456196",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'text': 'Wildlife  Meet the tuatara, ‘living fossil’ with fastest sperm in reptile world',\n",
       "  'link': 'https://www.theguardian.com/world/2021/aug/05/meet-the-tuatara-the-sluggish-living-fossil-with-the-fastest-sperm-in-the-reptile-world',\n",
       "  'domain': 'theguardian.com'},\n",
       " {'text': 'Bill Gates  Billionaire says meeting with Jeffrey Epstein was ‘a huge mistake’',\n",
       "  'link': 'https://www.theguardian.com/us-news/2021/aug/05/bill-gates-jeffrey-epstein-meeting-huge-mistake',\n",
       "  'domain': 'theguardian.com'},\n",
       " {'text': \"Woodland hermit's cabin fire leads to state investigation\",\n",
       "  'link': '//www.foxnews.com/us/new-hampshire-hermits-cabin-fire-investigation',\n",
       "  'domain': 'foxnews.com'},\n",
       " {'text': 'Thatcherism is the big Tory scam that still distorts our politics',\n",
       "  'link': 'https://www.theguardian.com/commentisfree/2021/aug/05/boris-johnson-thatcherism-state-taxes-high-earners-poor',\n",
       "  'domain': 'theguardian.com'},\n",
       " {'text': \"Don't miss out: Average mortgage refinance rates hold at 180-day low | August 5, 2021\",\n",
       "  'link': '//www.foxbusiness.com/personal-finance/todays-mortgage-refinance-rates-august-5-2021',\n",
       "  'domain': 'foxnews.com'},\n",
       " {'text': 'After decades in woods, New Hampshire man forced from cabin',\n",
       "  'link': '//www.foxnews.com/us/new-hampshire-man-forced-from-cabin-decades',\n",
       "  'domain': 'foxnews.com'},\n",
       " {'text': 'JPMorgan, Goldman Call Time on Work-From-Home. Their Rivals Are Ready to Pounce.',\n",
       "  'link': None,\n",
       "  'domain': 'wsj.com'},\n",
       " {'text': 'Tokyo Olympics',\n",
       "  'link': 'https://www.washingtonpost.com/sports/olympics/tokyo-summer-games/',\n",
       "  'domain': 'washingtonpost.com'},\n",
       " {'text': 'Strictly to have first all-male partnership',\n",
       "  'link': '/news/entertainment-arts-58089932',\n",
       "  'domain': 'bbc.com'},\n",
       " {'text': 'Troops Direct provides gear to service members in need',\n",
       "  'link': '//www.foxnews.com/us/troops-direct-nonprofit-service-members-gear',\n",
       "  'domain': 'foxnews.com'}]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from random import sample\n",
    "\n",
    "sample(titles, 10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6271aa8b",
   "metadata": {},
   "source": [
    "## Task Queue Concepts\n",
    "\n",
    "### Workers\n",
    "\n",
    "The workers carry out unit tasks of an end-to-end process. The tasks are dissected in a way that can run in isolation and incremental steps.\n",
    "\n",
    "In general, granular tasks are easier to implement, test, and in the context of a task queue, also more accessible to parallelize and maximize computing resources.\n",
    "\n",
    "### Queue\n",
    "\n",
    "The queue holds enqueued or scheduled tasks from arbitrary processes that do not require blocking (synchronous) execution. At the same time, workers spawned through other parallel computational means, such as processes, threads, or coroutines, dequeue and execute the tasks from the queue on a schedule or when there are available computing resources.\n",
    "\n",
    "### Communication protocol and broker\n",
    "\n",
    "Typical task queue design does not directly spawn processes (and subprocesses) to manage the queue data structure and tasks' operation (schedule and execution). It would be a struggle when dealing with the uncertainty of the complexity and scale of tasks and results.\n",
    "\n",
    "Instead, most task queues rely on a relatively agnostic communication broker to manage the queue data structure and a programming language agnostic serialization protocol (such as JSON) to transmit the task definitions and results through the queue. Some data stores are more suitable for tasks queues, categorized as _Message Queues_, specializing in inter-process communications.\n",
    "\n",
    "Such design also allows extensions, such as queue monitoring and scaling beyond a single machine, since most message queues provide standalone access and transmission across networks.\n",
    "\n",
    "![task-queue](https://user-images.githubusercontent.com/2837532/126832454-82a4a8e9-34a0-4ccc-9c39-83248a32be16.png)\n",
    "\n",
    "## Adapt Task Queue\n",
    "\n",
    "Let's extend the use-case and adapt a task queue to our advantage.\n",
    "\n",
    "From here, we will leverage another library called [Celery](https://docs.celeryproject.org/en/stable/index.html), a distributed task queue implementation. The Celery library handles the communication protocol and brokerage aspect of a task queue. It provides mature support for various queue backend and a simple interface for worker tasks definition and usage.\n",
    "\n",
    "In this part, we will use [Redis](https://redis.io/), an in-memory data store, as the backend for both the broker and results from worker tasks.\n",
    "\n",
    "Also, due to the nature of the task queue implementation, all task definitions would need to reside in standalone modules instead of notebook inline scripts. All task definitions can be found in [extract_tasks.py](extract_tasks.py)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "f702e2f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# the tasks reside in its own module\n",
    "import extract_tasks as tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0cdeba52",
   "metadata": {},
   "source": [
    "First, let's try a simple task that has a bare implementation as:\n",
    "\n",
    "```python\n",
    "@queue.task\n",
    "def add(a, b):\n",
    "    return a + b\n",
    "```\n",
    "\n",
    "Note the `@queue.task` on top of the function definition, a pattern known as a [decorator](https://wiki.python.org/moin/PythonDecorators), which is a technique that alters the default behaviour of a code block (in this case, function), without the need to change the original code block directly.\n",
    "\n",
    "The `@queue.task` is a decorator enabled by creating a Celery queue (or application) instance:\n",
    "\n",
    "```python\n",
    "from celery import Celery\n",
    "\n",
    "queue = Celery(\n",
    "    'news',  # name of the queue\n",
    "    broker='redis://localhost:6379/0',  # queue message broker backend\n",
    "    result_backend='redis://localhost:6379/1',  # task result backend\n",
    ")\n",
    "```\n",
    "\n",
    "By itself, the `add()` function would behave as expected:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "704825e7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tasks.add(5, 6)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18b53801",
   "metadata": {},
   "source": [
    "However, the decorator also grants it additional interfaces to behave like a worker task:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "7f0786cd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<AsyncResult: 678c5b92-3f91-4f41-861a-59dd0d4b358e>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = tasks.add.delay(5, 6)\n",
    "r"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "e766e6c9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'PENDING'"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.status"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5aabb38",
   "metadata": {},
   "source": [
    "At first, it may seem counter-intuitive to delay a task and have it in pending status after the invocation. The design is intended for the task queue to dispatch the particular task to an available worker to run asynchronously, thus not blocking the \"main\" process (in this case, the notebook we are in).\n",
    "\n",
    "At the moment, we still don't have any workers to take on the task. Therefore it would remain in the pending status:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "c5ccbdcf",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'PENDING'"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.status"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "79793c01",
   "metadata": {},
   "source": [
    "As mentioned before, the Celery project covers that part for us. All we have to do is to spawn a worker pool using its CLI (command-line-interface):\n",
    "\n",
    "```shell\n",
    "% celery --app extract_tasks worker --loglevel=INFO\n",
    "```\n",
    "\n",
    "We supply the command with the same module that holds the queue (application) and all the task definitions and a log-level at INFO (practically everything) to monitor all the details.\n",
    "\n",
    "As soon as we run the command, we would get something along the line of the following:\n",
    "\n",
    "```shell\n",
    "[2021-08-05 15:50:06,700: INFO/MainProcess] Task extract_tasks.add[678c5b92-3f91-4f41-861a-59dd0d4b358e] received\n",
    "[2021-08-05 15:50:06,724: INFO/ForkPoolWorker-1] Task extract_tasks.add[678c5b92-3f91-4f41-861a-59dd0d4b358e] succeeded in 0.01997902699999976s: 11\n",
    "```\n",
    "\n",
    "The worker `MainProcess` receives tasks from anywhere that invokes them, such as this notebook instance. Then an available `Worker-N` of a specific type (in this case, a multi-processing \"forked\" process) would take on the task and execute it. The information log indicates that the task succeeded, with a returned result of `11`.\n",
    "\n",
    "In this case, it would be somewhat meaningless if we cannot get the intended result. That's what the result backend serves its purpose, and Celery provides a straightforward interface for us to obtain it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "b9779e59",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'SUCCESS'"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "f4173f8b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.get()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1ece7a9",
   "metadata": {},
   "source": [
    "Let's incorporate the task version of the `fetch` function through the task queue:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "b6e25440",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 222.83it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 11.5 ms, sys: 4.28 ms, total: 15.8 ms\n",
      "Wall time: 26.5 ms\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'bbc.com': <AsyncResult: ba871623-b104-4c17-9b00-bf8b9660b252>,\n",
       " 'theguardian.com': <AsyncResult: dab084d7-c483-4220-a8ff-0c5492e243de>,\n",
       " 'washingtonpost.com': <AsyncResult: 2f81e454-b7da-4fa6-93e3-7eb498a39f22>,\n",
       " 'foxnews.com': <AsyncResult: ca14d847-7a37-4144-ac2f-e47ae2520251>,\n",
       " 'wsj.com': <AsyncResult: 39f019b7-3b21-4068-8236-0a7d095cf026>}"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "sites = [\n",
    "    'bbc.com',\n",
    "    'theguardian.com',\n",
    "    'washingtonpost.com',\n",
    "    'foxnews.com',\n",
    "    'wsj.com',\n",
    "]\n",
    "data = {}\n",
    "for domain in tqdm(sites, ncols=100):\n",
    "    data[domain] = tasks.fetch_website_task.delay(domain)\n",
    "\n",
    "data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3840611",
   "metadata": {},
   "source": [
    "Naturally, unlike the iterative and blocking version from the beginning, the above snippet merely enqueues the tasks and responds with the `AsyncResult` object, as seen with the `add` task case, which means we can query the status and potentially get its result on success:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "48dfcd63",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "  <!doctype html><!--GRAND CANYON PREBID -->\n",
      "  <html lang=en>\n",
      "    <head>\n",
      "      <meta charSet='utf-8' />\n",
      "      <meta name=\"description\" content=\"We can’t find the page you are looking for.\"/>\n",
      "    <meta name=\"keywords\" content=\"\" />\n",
      "    <meta name=\"page.section\" content=\"Error\" />\n",
      "    <meta name=\"pag\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "from pprint import pprint\n",
    "\n",
    "while data['wsj.com'].status == 'PENDING':\n",
    "    time.sleep(0.01)\n",
    "\n",
    "print(data['wsj.com'].get()['raw'][:300])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20505b61",
   "metadata": {},
   "source": [
    "From above, we use a `while` loop with a condition on `.status == 'PENDING'` to emulate the synchronous and blocking operation. After observing the task has succeeded, we then obtain the result using the `.get()` method. We can verify this by observing the worker logs:\n",
    "\n",
    "```shell\n",
    "[2021-08-05 17:12:39,554: INFO/MainProcess] Task extract_tasks.fetch_website_task[ba871623-b104-4c17-9b00-bf8b9660b252] received\n",
    "[2021-08-05 17:12:39,556: INFO/MainProcess] Task extract_tasks.fetch_website_task[dab084d7-c483-4220-a8ff-0c5492e243de] received\n",
    "[2021-08-05 17:12:39,558: INFO/MainProcess] Task extract_tasks.fetch_website_task[2f81e454-b7da-4fa6-93e3-7eb498a39f22] received\n",
    "[2021-08-05 17:12:39,562: INFO/MainProcess] Task extract_tasks.fetch_website_task[ca14d847-7a37-4144-ac2f-e47ae2520251] received\n",
    "[2021-08-05 17:12:39,575: INFO/MainProcess] Task extract_tasks.fetch_website_task[39f019b7-3b21-4068-8236-0a7d095cf026] received\n",
    "[2021-08-05 17:12:40,356: INFO/ForkPoolWorker-8] Task extract_tasks.fetch_website_task[ba871623-b104-4c17-9b00-bf8b9660b252] succeeded in 0.7991691030000005s: <...>\n",
    "[2021-08-05 17:12:40,510: INFO/ForkPoolWorker-2] Task extract_tasks.fetch_website_task[ca14d847-7a37-4144-ac2f-e47ae2520251] succeeded in 0.9463052399999974s: <...>\n",
    "[2021-08-05 17:12:40,880: INFO/ForkPoolWorker-10] Task extract_tasks.fetch_website_task[39f019b7-3b21-4068-8236-0a7d095cf026] succeeded in 1.3033900759999995s: <...>\n",
    "[2021-08-05 17:12:40,919: INFO/ForkPoolWorker-1] Task extract_tasks.fetch_website_task[dab084d7-c483-4220-a8ff-0c5492e243de] succeeded in 1.361990458000001s: <...>\n",
    "[2021-08-05 17:12:50,535: INFO/ForkPoolWorker-9] Task extract_tasks.fetch_website_task[2f81e454-b7da-4fa6-93e3-7eb498a39f22] succeeded in 10.970096680999998s: <...>\n",
    "```\n",
    "\n",
    "A few observations:\n",
    "1. The task IDs correlate precisely like the ones after the `AsyncResult` object. The receiving order of tasks follows precisely as the for loop iteration dictated.\n",
    "2. There are differently numbered workers that took on the tasks.\n",
    "3. Both workers and tasks are paired out-of-order, naturally so are the completions."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "49b2d984",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Pipelining\n",
    "\n",
    "Task queue implementations sometimes come with pipelining (or chaining) capability to arrange unit tasks as a pipeline with a specific logical order for data to flow through them. The scheduling mechanism manages available workers to take on unit tasks as they become available (as demonstrated as an out-of-order pairing between workers and tasks) while maintaining data flow between unit tasks in a given pipeline.\n",
    "\n",
    "Let's compose such a pipeline chain using the unit tasks from [extract_tasks.py](extract_tasks.py) to complete the workflow to extract headline title text and links as we have done before without the task queue:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "32b86745",
   "metadata": {},
   "outputs": [],
   "source": [
    "def pipe(domain):\n",
    "    chain = tasks.fetch_website_task.s(domain) | tasks.extract_titles.s()\n",
    "    return chain()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a03e9675",
   "metadata": {},
   "source": [
    "The `.s()` method, granted by the `@queue.task` decorator like `.delay()`, generates the function \"signature\" for the task worker to understand how to execute it. The pipe operator `|` pipes the returned value from left to right, in such order.\n",
    "\n",
    "The resulting `chain()` returns an `AsyncResult` object, just like when we invoke individual task functions with the decorated `.delay()` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "2e646c10",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<AsyncResult: 6de32df7-6702-4635-ab9b-ddb1cc1cda22>"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = pipe(domain='bbc.com')\n",
    "r"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "0576960d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'SUCCESS'"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "cdbecc58",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'text': 'Belarus athlete told by grandmother not to return',\n",
       "  'link': 'https://bbc.com/news/world-europe-58104195'},\n",
       " {'text': 'Chronic illness influencers accused of faking it',\n",
       "  'link': 'https://bbc.com/news/stories-58093455'},\n",
       " {'text': 'Ethiopian rebels take Unesco world heritage town',\n",
       "  'link': 'https://bbc.com/news/world-africa-58101912'},\n",
       " {'text': \"Turkish influencer prosecuted 'for sex-toy photos'\",\n",
       "  'link': 'https://bbc.com/news/world-europe-58102368'},\n",
       " {'text': 'Man City sign £100m Grealish from Villa',\n",
       "  'link': 'https://bbc.com/sport/football/57818660'},\n",
       " {'text': 'Mixed day for USA on day 13 in Tokyo',\n",
       "  'link': 'https://bbc.com/sport/olympics/58096833'},\n",
       " {'text': 'Anderson leads England fightback',\n",
       "  'link': 'https://bbc.com/sport/cricket/58106765'},\n",
       " {'text': \"Olympic athlete: 'It's dangerous for me in Belarus'\",\n",
       "  'link': 'https://bbc.com/news/world-europe-58099987'},\n",
       " {'text': \"Olympic athlete: 'It's dangerous for me in...\",\n",
       "  'link': 'https://bbc.com/news/world-europe-58099987'},\n",
       " {'text': \"'You can be African and black and play polo'\",\n",
       "  'link': 'https://bbc.com/news/uk-58049010'}]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r.get()['titles'][:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "927dbc77",
   "metadata": {},
   "source": [
    "Let's replay through all the websites with the pipeline version:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "8f2b5c71",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 387.21it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 13.6 ms, sys: 3.7 ms, total: 17.3 ms\n",
      "Wall time: 17.3 ms\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'bbc.com': <AsyncResult: 667ceb82-8b0c-4af0-8edf-1c01e1748a40>,\n",
       " 'theguardian.com': <AsyncResult: c4a3c7a3-3184-4daf-a336-72674006b70b>,\n",
       " 'washingtonpost.com': <AsyncResult: 1be94109-51a0-488e-966f-e4b8c41f3f04>,\n",
       " 'foxnews.com': <AsyncResult: a9a953d2-b455-4bce-bf3d-8a19518feed8>,\n",
       " 'wsj.com': <AsyncResult: 89f08eb7-dbbd-4f93-970f-21fe8bf36deb>}"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "sites = [\n",
    "    'bbc.com',\n",
    "    'theguardian.com',\n",
    "    'washingtonpost.com',\n",
    "    'foxnews.com',\n",
    "    'wsj.com',\n",
    "]\n",
    "data = {}\n",
    "for domain in tqdm(sites, ncols=100):\n",
    "    data[domain] = pipe(domain)\n",
    "\n",
    "data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "024cc5fb",
   "metadata": {
    "tags": []
   },
   "source": [
    "And observe the worker logs:\n",
    "\n",
    "```shell\n",
    "[2021-08-05 17:26:10,319: INFO/MainProcess] Task extract_tasks.fetch_website_task[10f709fd-b350-4729-973f-d04ef7f1415a] received\n",
    "[2021-08-05 17:26:10,323: INFO/MainProcess] Task extract_tasks.fetch_website_task[92ebcebd-2853-406d-b44f-97ed23fc6454] received\n",
    "[2021-08-05 17:26:10,325: INFO/MainProcess] Task extract_tasks.fetch_website_task[87f036c6-fbff-45d3-a14a-a6dfe17cf942] received\n",
    "[2021-08-05 17:26:10,328: INFO/MainProcess] Task extract_tasks.fetch_website_task[cda76c38-1e09-4267-aaef-9816787ddc97] received\n",
    "[2021-08-05 17:26:10,331: INFO/MainProcess] Task extract_tasks.fetch_website_task[f43a7963-ac43-4afe-b962-be469b7075e2] received\n",
    "[2021-08-05 17:26:10,568: INFO/ForkPoolWorker-8] Task extract_tasks.fetch_website_task[10f709fd-b350-4729-973f-d04ef7f1415a] succeeded in 0.24766298300005474s: <...>\n",
    "[2021-08-05 17:26:10,576: INFO/MainProcess] Task extract_tasks.extract_titles[667ceb82-8b0c-4af0-8edf-1c01e1748a40] received\n",
    "[2021-08-05 17:26:10,671: INFO/ForkPoolWorker-8] Task extract_tasks.extract_titles[667ceb82-8b0c-4af0-8edf-1c01e1748a40] succeeded in 0.08766871499994977s: <...>\n",
    "[2021-08-05 17:26:10,722: INFO/ForkPoolWorker-10] Task extract_tasks.fetch_website_task[f43a7963-ac43-4afe-b962-be469b7075e2] succeeded in 0.38966748499990445s: <...>\n",
    "[2021-08-05 17:26:10,741: INFO/MainProcess] Task extract_tasks.extract_titles[89f08eb7-dbbd-4f93-970f-21fe8bf36deb] received\n",
    "[2021-08-05 17:26:10,765: INFO/ForkPoolWorker-2] Task extract_tasks.fetch_website_task[cda76c38-1e09-4267-aaef-9816787ddc97] succeeded in 0.43293487899995853s: <...>\n",
    "[2021-08-05 17:26:10,770: INFO/ForkPoolWorker-1] Task extract_tasks.fetch_website_task[92ebcebd-2853-406d-b44f-97ed23fc6454] succeeded in 0.44436259199994765s: <...>\n",
    "[2021-08-05 17:26:10,779: INFO/MainProcess] Task extract_tasks.extract_titles[c4a3c7a3-3184-4daf-a336-72674006b70b] received\n",
    "[2021-08-05 17:26:10,795: INFO/MainProcess] Task extract_tasks.extract_titles[a9a953d2-b455-4bce-bf3d-8a19518feed8] received\n",
    "[2021-08-05 17:26:10,923: INFO/ForkPoolWorker-8] Task extract_tasks.extract_titles[89f08eb7-dbbd-4f93-970f-21fe8bf36deb] succeeded in 0.1257382599999346s: <...>\n",
    "[2021-08-05 17:26:11,128: INFO/ForkPoolWorker-2] Task extract_tasks.extract_titles[a9a953d2-b455-4bce-bf3d-8a19518feed8] succeeded in 0.3050345090000519s: <...>\n",
    "[2021-08-05 17:26:11,221: INFO/ForkPoolWorker-1] Task extract_tasks.extract_titles[c4a3c7a3-3184-4daf-a336-72674006b70b] succeeded in 0.3850136209999846s: <...>\n",
    "[2021-08-05 17:26:20,781: INFO/ForkPoolWorker-9] Task extract_tasks.fetch_website_task[87f036c6-fbff-45d3-a14a-a6dfe17cf942] succeeded in 10.455224110000017s: <...>\n",
    "[2021-08-05 17:26:20,796: INFO/MainProcess] Task extract_tasks.extract_titles[1be94109-51a0-488e-966f-e4b8c41f3f04] received\n",
    "[2021-08-05 17:26:20,880: INFO/ForkPoolWorker-8] Task extract_tasks.extract_titles[1be94109-51a0-488e-966f-e4b8c41f3f04] succeeded in 0.07039165100002265s: <...>\n",
    "```\n",
    "\n",
    "It may be puzzling at first, but this behaviour matches our expectation from the previously single-task observation, with some subtle differences:\n",
    "1. The order of tasks and task IDs exhibit some differences:\n",
    "    a. The receiving order of the `fetch_website_task` tasks follow precisely as the for loop iteration dictated, but not the `extract_titles` tasks.\n",
    "    b. The `AsyncResult` IDs correlate to the IDs of the `extract_titles` tasks.\n",
    "2. There are differently numbered workers that took on the tasks.\n",
    "3. The pairing of workers and tasks is also an interesting observation:\n",
    "    a. Both workers and tasks are paired out-of-order, naturally so are the completions.\n",
    "    b. The queue would only dispatch a `extract_titles` task after its predecessor task (`fetch_website_task`) of the same pipeline/chain succeeds.\n",
    "    c. The same worker can take on the same or different types of unit tasks, and from each of the worker's perspectives, does not care about the order of the unit tasks execution (the queue manages the order).\n",
    "    \n",
    "Workers still pair and execute unit tasks out-of-order while the queue seamlessly assembles the result back in order, further empowers task queues to take on more complex applications.\n",
    "\n",
    "## Server Application\n",
    "\n",
    "One of the most appealing traits of task queue is that it offers asynchronous exit for the \"main\" process, which makes it a valuable adoption for web services to defer and execute potentially slow tasks in the backend.\n",
    "\n",
    "First, let's build the server out using the [Flask](https://flask.palletsprojects.com/en/2.0.x/) web framework for Python, by adapting the previous `pipe()` function:\n",
    "\n",
    "```python\n",
    "from urllib.parse import urlparse\n",
    "\n",
    "from flask import Flask, jsonify, request\n",
    "\n",
    "from extract_tasks import (\n",
    "    filter_rules,\n",
    "    fetch_website_task,\n",
    "    extract_titles,\n",
    ")\n",
    "\n",
    "\n",
    "app = Flask(__name__)\n",
    "\n",
    "\n",
    "def pipe(domain):\n",
    "    chain = fetch_website_task.s(domain) | extract_titles.s()\n",
    "    return chain()\n",
    "\n",
    "\n",
    "@app.route('/', methods=['GET', 'POST'])\n",
    "def get_news():\n",
    "    domain = urlparse(request.args.get('domain'))\n",
    "    domain = domain.netloc or domain.path or 'bbc.com'\n",
    "\n",
    "    if domain not in filter_rules:\n",
    "        return jsonify({\n",
    "            'error': f'{domain} not supported. it needs to be one of {\", \".join(filter_rules.keys())}',\n",
    "        })\n",
    "\n",
    "    # enqueue the task while not blocking the server to immediately respond to the requesting client\n",
    "    pipe(domain, response_url)\n",
    "    return jsonify({\n",
    "        'text': f'parsing {domain} headlines',\n",
    "    })\n",
    "```\n",
    "\n",
    "Like the task queue implementation, the server application would also need to reside in its module, found in [server.py](./server.py). Just like the Celery queue implementation, we can also launch a Flask application server using its CLI:\n",
    "\n",
    "```shell\n",
    "% FLASK_APP=server FLASK_DEBUG=1 flask run      \n",
    " * Serving Flask app 'server' (lazy loading)\n",
    " * Environment: production\n",
    "   WARNING: This is a development server. Do not use it in a production deployment.\n",
    "   Use a production WSGI server instead.\n",
    " * Debug mode: on\n",
    " * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)\n",
    " * Restarting with stat\n",
    " * Debugger is active!\n",
    " * Debugger PIN: 798-476-896\n",
    "```\n",
    "\n",
    "The string `FLASK_APP=server FLASK_DEBUG=1` before the actual command `flask run` defines two environment variables to control the command behaviour. This is a different approach than what the Celery CLI uses with the options such as `--app` and `--loglevel`, but ultimately achieving the same end-goals. The debug mode of Flask also enables \"hot-reloading\" of the server, making it convenient to modify our server code without restarting it.\n",
    "\n",
    "Once the server starts, we can access it through our \"localhost\" (or the IP address `127.0.0.1`) at the specified port `:5000`:\n",
    "\n",
    "![server](https://user-images.githubusercontent.com/2837532/128428104-3c405b69-2cd3-4da8-9742-13520f2b0eec.png)\n",
    "\n",
    "The deference of tasks also means that the server can handle much more traffic than otherwise, especially if we leverage additional server nodes to function as task queue workers.\n",
    "\n",
    "### Where do we send the results?\n",
    "\n",
    "While the task queue allows the server to respond immediately, there is also a substantial (and obvious) trade-off -- the requesting client-side would not receive the requested information unless we have a place to deliver the deferred results.\n",
    "\n",
    "The usual delivery destinations can be an email, SMS (Short Message Service), and increasingly group chats, such as Slack. So let's build a companion slash command `/news` in our Slack workspace to request and obtain news headlines by modifying the pipe function to include the task to post to Slack and the server handler to expect the specific requesting payload from a Slack slash command.\n",
    "\n",
    "![pipeline-pool](https://user-images.githubusercontent.com/2837532/128397170-aec0a57f-104f-4407-86ff-a3a8d5ba301f.png)\n",
    "\n",
    "```python\n",
    "# ...previous imports...\n",
    "\n",
    "from extract_tasks import post_slack\n",
    "\n",
    "def pipe_v2(domain, response_url):\n",
    "    chain = fetch_website_task.s(domain) | extract_titles.s() | post_slack.s(response_url)\n",
    "    return chain()\n",
    "\n",
    "\n",
    "@app.route('/', methods=['GET', 'POST'])\n",
    "def get_news():\n",
    "    # request.form\n",
    "    # command - the slash command\n",
    "    # text - after command\n",
    "    # response_url - the POST back URL for follow up messages\n",
    "    domain = urlparse(request.form.get('text') or request.args.get('domain'))\n",
    "    domain = domain.netloc or domain.path or 'bbc.com'\n",
    "\n",
    "    if domain not in filter_rules:\n",
    "        return jsonify({\n",
    "            'error': f'{domain} not supported. it needs to be one of {\", \".join(filter_rules.keys())}',\n",
    "        })\n",
    "\n",
    "    # queue up the task\n",
    "    response_url = request.form.get('response_url')\n",
    "    if not response_url:\n",
    "        pipe(domain)\n",
    "    else:\n",
    "        pipe_v2(domain, response_url)\n",
    "\n",
    "    return jsonify({\n",
    "        'text': f'parsing {domain} headlines',\n",
    "    })\n",
    "```\n",
    "\n",
    "Once we define the slash command in an installed Slack App and direct its \"Request URL\" to our Flask server, we are all set to get news headlines in Slack by typing `/news`:\n",
    "\n",
    "![slash](https://user-images.githubusercontent.com/2837532/128430462-7ae10cf2-9d90-4fc5-a383-8aa7c0e5024c.png)\n",
    "\n",
    "![slash-request](https://user-images.githubusercontent.com/2837532/128430738-f1d927a8-471f-4a0f-8d13-a93eed3dfd09.png)\n",
    "\n",
    "And upon the post-task completion, we will receive a follow-up of all the headlines delivered to where we requested them:\n",
    "\n",
    "![slack-response](https://user-images.githubusercontent.com/2837532/128430856-cec456de-dea4-4c99-9fe6-3749c53554f2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f52f9d2-c232-4178-bb5e-c779f7fa13a8",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Remarks\n",
    "\n",
    "Task (or job) queues can often fit in rather complex software, and as a result, offer some significant benefits:\n",
    "\n",
    "* Simplify the software architecture\n",
    "* Improve the utilization of computing resources\n",
    "* Improve the scalability of the software\n",
    "\n",
    "## References\n",
    "\n",
    "* [Part 14 - Distributed Computation](../14-distributed-computation/notebook.ipynb)\n",
    "* [Beautiful Soup](https://www.crummy.com/software/BeautifulSoup/bs4/doc/)\n",
    "* [Celery](https://docs.celeryproject.org/en/stable/index.html)\n",
    "* [Flask](https://flask.palletsprojects.com/en/2.0.x/)\n",
    "* [tqdm](https://tqdm.github.io/)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}