3 Something more than simple tasks.ipynb

{
 "metadata": {
  "name": ""
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "![](https://github.com/PyCon/2014/blob/master/assets/logos/pycon2014-blog-image.png?raw=true)\n",
      "Distibuted task processing using Celery\n",
      "======================================="
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Create some positions in Redis - we will access these from the Celery tasks later"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import redis\n",
      "import simplejson as json\n",
      "\n",
      "#the database\n",
      "db=redis.Redis()\n",
      "\n",
      "#the positions\n",
      "positions=[\n",
      "{'name':'pos_1','size':100,'price':0.99},\n",
      "{'name':'pos_2','size':100,'price':0.91},\n",
      "{'name':'pos_3','size':100,'price':0.85},\n",
      "{'name':'pos_4','size':100,'price':0.96},\n",
      "{'name':'pos_5','size':100,'price':0.79}\n",
      "]\n",
      "\n",
      "def create_positions():\n",
      "    for position in positions:\n",
      "        db.set(position['name'],json.dumps(position))\n",
      "      \n",
      "\n"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "create_positions()\n",
      "print json.loads(db.get('pos_1'))\n",
      "db.delete('pos_123')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "{'price': 0.99, 'name': 'pos_1', 'size': 100}\n"
       ]
      },
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 2,
       "text": [
        "0"
       ]
      }
     ],
     "prompt_number": 2
    },
    {
     "cell_type": "raw",
     "metadata": {},
     "source": [
      "import simplejson as json\n",
      "from celery import Celery\n",
      "from celery.utils.log import get_task_logger\n",
      "import redis\n",
      "\n",
      "celery=Celery('positions',broker='redis://',backend='redis://')\n",
      "logger = get_task_logger(__name__)\n",
      "\n",
      "@celery.task\n",
      "def get_initial_position_value(name):\n",
      "    db=redis.Redis()\n",
      "    pos=json.loads(db.get(name))\n",
      "    return pos['size']*pos['price']\n",
      "    \n",
      "\n",
      "@celery.task\n",
      "def get_value_at_close(name,close):\n",
      "    db=redis.Redis()\n",
      "    pos=json.loads(db.get(name))\n",
      "    return pos['size']*close\n",
      "    \n",
      "@celery.task\n",
      "def get_value_at_close_1(name,close):\n",
      "    db=redis.Redis()\n",
      "    pos=json.loads(db.get(name))\n",
      "    logger.info('Calculating value with close price %s'%(close,))\n",
      "    return pos['size']*close\n"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from eval_positions import *\n",
      "t1=get_initial_position_value.delay('pos_1')\n",
      "t2=get_value_at_close.delay('pos_1',0.992)\n",
      "t3=get_value_at_close.delay('pos_1',0.992)\n",
      "t4=get_initial_position_value.delay('pos_123')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print t1.state, t1.result\n",
      "print t2.state, t2.result\n",
      "print t3.state, t3.result\n",
      "print t4.state, t4.result"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "SUCCESS 99.0\n",
        "SUCCESS 99.2\n",
        "SUCCESS 99.2\n",
        "FAILURE expected string or buffer\n"
       ]
      }
     ],
     "prompt_number": 4
    },
    {
     "cell_type": "raw",
     "metadata": {},
     "source": [
      "import simplejson as json\n",
      "from celery import Celery\n",
      "from celery.utils.log import get_task_logger\n",
      "import redis\n",
      "\n",
      "celery=Celery('positions',broker='redis://',backend='redis://')\n",
      "logger = get_task_logger(__name__)\n",
      "\n",
      "@celery.task\n",
      "def get_initial_position_value(name):\n",
      "    db=redis.Redis()\n",
      "    pos=json.loads(db.get(name))\n",
      "    if pos is None:\n",
      "        return\n",
      "    else:\n",
      "        return pos['size']*pos['price']\n",
      "    \n",
      "\n",
      "@celery.task\n",
      "def get_value_at_close(name,close):\n",
      "    if close is None or close<0:\n",
      "        return None\n",
      "    db=redis.Redis()\n",
      "    pos_str=db.get(name)\n",
      "    if pos_str is None:\n",
      "        return None\n",
      "    else:\n",
      "        pos=json.loads(pos_str)\n",
      "        return pos['size']*close\n",
      "    \n",
      "@celery.task\n",
      "def get_value_at_close_retry(name,close):\n",
      "    if close is None or close<0:\n",
      "        return None\n",
      "    db=redis.Redis()\n",
      "    pos_str=db.get(name)\n",
      "    if pos_str is None:\n",
      "        raise get_value_at_close_retry.retry()\n",
      "    else:\n",
      "        pos=json.loads(pos_str)\n",
      "        return pos['size']*close\n"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import eval_positions_1\n",
      "t4=eval_positions_1.get_value_at_close_retry.delay('pos_123',0.9945)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 5
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print t4.state"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "RETRY\n"
       ]
      }
     ],
     "prompt_number": 17
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "p={'name':'pos_1','size':100,'price':0.99}\n",
      "db.set('pos_123',json.dumps(p))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 18,
       "text": [
        "True"
       ]
      }
     ],
     "prompt_number": 18
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print t4.state"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "SUCCESS\n"
       ]
      }
     ],
     "prompt_number": 40
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print t4.result"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "99.45\n"
       ]
      }
     ],
     "prompt_number": 42
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print db.get('pos_123')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "{\"price\": 0.99, \"name\": \"pos_1\", \"size\": 100}\n"
       ]
      }
     ],
     "prompt_number": 37
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "\n",
      "- The task retry does not send to the same worker again - so you need to start another worker now and then you will see that the task has been retried. \n",
      "- There is a way around this - create queues for tasks, not the default queue whcih we have been using so far, and then pass the queue name to the retry function\n",
      "\n",
      "task.retry(queue='queue name')"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "- more on retrying - http://docs.celeryproject.org/en/latest/userguide/tasks.html#retrying"
     ]
    }
   ],
   "metadata": {}
  }
 ]
}