Intro to async Python and Sanic

Async+sanic.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Asynchronous programming in Python\n",
    "\n",
    "Here are my first steps into asynchronous programming in Python. If you are completely new to the subject, I would recommend [How the heck does async/await work in Python 3.5?](https://snarky.ca/how-the-heck-does-async-await-work-in-python-3-5/).\n",
    "\n",
    "Let's say I want to execute 5 tasks. Each task has to wait 10 seconds. In a synchronous world, the overall duration would be 5 * 10 = 50 seconds:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0: going to sleep at 0\n",
      "0: wake up at 10\n",
      "1: going to sleep at 10\n",
      "1: wake up at 20\n",
      "2: going to sleep at 20\n",
      "2: wake up at 30\n",
      "3: going to sleep at 30\n",
      "3: wake up at 40\n",
      "4: going to sleep at 40\n",
      "4: wake up at 50\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "def do_sleep(task_id, start=0):\n",
    "    print('{}: going to sleep at {}'.format(task_id, int(time.time() - start)))\n",
    "    time.sleep(10)\n",
    "    print('{}: wake up at {}'.format(task_id, int(time.time() - start)))\n",
    "\n",
    "start = time.time()\n",
    "for i in range(5):\n",
    "    do_sleep(i, start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "With Python3.5 and `asyncio` you could write the `do_sleep` task as a *coroutine*. For a proper explanation of what's considered to be a coroutine in Python, check out the blog post I referred to earlier. For now, it suffices to think of a coroutine as a function who's execution can be paused - handing the control back to the caller - and resumed at given moments.\n",
    "\n",
    "A coroutine in Python is defined with the `async def` syntax. In such a coroutine, a `await other_coroutine()` statement can exist. At this point, the execution of the coroutine can be paused and depending on what `other_coroutine` does, it can hand the control back to the caller. The `asyncio.sleep()` coroutine is such an example.\n",
    "\n",
    "Here is how our task would look like as a coroutine:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "\n",
    "async def do_sleep_async(task_id, start=0):\n",
    "    print('{}: going to sleep at {}'.format(task_id, int(time.time() - start)))\n",
    "    await asyncio.sleep(10)\n",
    "    print('{}: wake up at {}'.format(task_id, int(time.time() - start)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "A final concept that I need to explain very briefly is the *event loop*. Basically, the event loop manages the execution of coroutines. You can schedule the execution of a coroutine by creating a task on the event loop and once you start running the event loop, it will start executing the coroutines.\n",
    "\n",
    "In the next code block, I create an event loop, create 5 tasks, and tell the event loop to run until all tasks are complete. I'm not sure whether this is the best way to schedule a list of tasks, but the next example works:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0: going to sleep at 0\n",
      "1: going to sleep at 0\n",
      "2: going to sleep at 0\n",
      "3: going to sleep at 0\n",
      "4: going to sleep at 0\n",
      "0: wake up at 10\n",
      "1: wake up at 10\n",
      "2: wake up at 10\n",
      "3: wake up at 10\n",
      "4: wake up at 10\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "my_event_loop = asyncio.get_event_loop()\n",
    "\n",
    "tasks = []\n",
    "for i in range(5):\n",
    "    task = my_event_loop.create_task(do_sleep_async(i, start))\n",
    "    tasks.append(task)\n",
    "\n",
    "my_event_loop.run_until_complete(asyncio.gather(*tasks))\n",
    "my_event_loop.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Sanic - a Python web framework based on coroutines\n",
    "\n",
    "Having a web framework that uses coroutines to handle incoming requests can reduce your response times. After all, while your server is requesting data from a different service to handle one request, it can already start processing the next request.\n",
    "\n",
    "Furthermore, we can use the asynchronous execution model to start background jobs. The particular use case I was interested in is the case where the server returns a job id and a url at wich the caller can start polling for his results. The server executes (or offloads) the job in the background and when it is ready, the caller can get the results at a different url.\n",
    "\n",
    "Let's say I want to download biological data from a data provider, but preparing the download takes some time for the provider. Here is how that interaction could look like:\n",
    "\n",
    "- Me: *\"I want a download for these and these filters\"*\n",
    "- Server: *\"Your job id is: job-id\"*\n",
    "- Me: *\"How is job job-id going along?\"*\n",
    "- Server: *\"Working on it\"*\n",
    "- wait some time\n",
    "- Me: *\"How is job job-id going along?\"*\n",
    "- Server: *\"Working on it\"*\n",
    "- wait some time\n",
    "- Me: *\"How is job job-id going along?\"*\n",
    "- Server: *\"Working on it\"*\n",
    "- wait some time\n",
    "- Me: *\"How is job job-id going along?\"*\n",
    "- Server: *\"It is done. You can get the results at this-url\"*\n",
    "- Me: *\"Download files at this-url\"*\n",
    "\n",
    "Let's start with a synchronous implementation of the download-data use case. And instead of actually preparing the download, I let the `prepare_download` function go to sleep for 5 seconds."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "from sanic import Sanic\n",
    "from sanic.response import json\n",
    "\n",
    "app = Sanic()\n",
    "\n",
    "def prepare_download(start=0):\n",
    "    print('waiting at {}'.format(int(time.time() - start)))\n",
    "    time.sleep(5)\n",
    "    print('done waiting at {}'.format(int(time.time() - start)))\n",
    "\n",
    "@app.route('/download')\n",
    "def test(request):\n",
    "    start = time.time()\n",
    "    print('start: {}'.format(int(time.time() - start)))\n",
    "    prepare_download(start)\n",
    "    print('stop: {}'.format(int(time.time() - start)))\n",
    "    return json({'ok': 1})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "If I run that, and visit `localhost:5000/download`, the output in my servers console is:\n",
    "\n",
    "```\n",
    "start: 0\n",
    "waiting at 0\n",
    "done waiting at 5\n",
    "stop: 5\n",
    "```\n",
    "\n",
    "So the request came in at the server, it called `prepare_download` and waited for that function to finish before returning a response to the caller.\n",
    "\n",
    "Now what we want to achieve is that we can return the job id immediately to the caller, and schedule the `prepare_download` task in the background. Here is how we can do that using Python's coroutines:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "async def prepare_download_async(start=0):\n",
    "    print('waiting at {}'.format(int(time.time() - start)))\n",
    "    await asyncio.sleep(5)\n",
    "    print('done waiting at {}'.format(int(time.time() - start)))\n",
    "\n",
    "@app.route('/async_test')\n",
    "async def test_async(request):\n",
    "    start = time.time()\n",
    "    print('start: {}'.format(int(time.time() - start)))\n",
    "    asyncio.ensure_future(prepare_download_async(start))\n",
    "    print('stop: {}'.format(int(time.time() - start)))\n",
    "    return json({'ok': 1})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "The output in the console when I now visit `localhost:5000/async_test` is:\n",
    "\n",
    "```\n",
    "start: 0\n",
    "stop: 0\n",
    "waiting at 0\n",
    "done waiting at 5\n",
    "```\n",
    "\n",
    "So the caller function `test_async` did not wait for `prepare_download_async` to finish. That is achieved using `asyncio.ensure_future()`.\n",
    "\n",
    "It is now a small step towards an implementation where the caller gets a job id, and can poll for the results of that job. For demonstration purposes, we'll store the job status and results in a global dictionary. One can imagine better solutions for that."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import uuid\n",
    "from sanic import Sanic\n",
    "from sanic.response import json\n",
    "\n",
    "app = Sanic()\n",
    "\n",
    "job_statuses = {}\n",
    "job_results = {}\n",
    "\n",
    "async def prepare_download(job_id):\n",
    "    await asyncio.sleep(20)\n",
    "    job_results[job_id] = 'the result of the job'\n",
    "    job_statuses[job_id] = 'success'\n",
    "\n",
    "\n",
    "@app.route('/download')\n",
    "async def request_download(request):\n",
    "    \"\"\"request a download\"\"\"\n",
    "    job_id = str(uuid.uuid4())\n",
    "    job_statuses[job_id] = 'pending'\n",
    "    asyncio.ensure_future(prepare_download(job_id))\n",
    "    return json({'job_id': job_id})\n",
    "\n",
    "\n",
    "@app.route('/job/<job_id>')\n",
    "async def get_job_status(request, job_id):\n",
    "    \"\"\"check a jobs status\"\"\"\n",
    "    if job_id in job_statuses:\n",
    "        return json({job_id: job_statuses[job_id]})\n",
    "    else:\n",
    "        return json({job_id: 'unknown job id'})\n",
    "\n",
    "@app.route('/results/<job_id>')\n",
    "async def get_job_results(request, job_id):\n",
    "    \"\"\"get results of a job\"\"\"\n",
    "    if job_id in job_results:\n",
    "        return json({job_id: job_results[job_id]})\n",
    "    else:\n",
    "        return json({job_id: 'no results'})"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}