docker/aiohttp demo (#75)

* adding docker/aiohttp demo

* adding aiohttp docs, fix #69

* add history

Author: samuelcolvin

.gitignore

@@ -1,15 +1,15 @@
-diff-match-patch
-env
-.idea
+/env
+/.idea
 __pycache__/
 *.py[cod]
 *.cache
 .coverage.*
-.coverage
-htmlcov/
+/.coverage
+/htmlcov/
 /build
 /dist
-test.py
+/test.py
 *.egg-info
-docs/_build/
-.mypy_cache/
+/docs/_build/
+/.mypy_cache/
+/demo/tmp/

HISTORY.rst

@@ -3,10 +3,11 @@
 History
 -------
 
-v0.12.0 (2017-11-XX)
+v0.12.0 (2017-11-16)
 ....................
 * better signal handling, support ``uvloop`` #73
 * drain pending tasks and drain task cancellation #74
+* add aiohttp and docker demo ``/demo`` #75
 
 v0.11.0 (2017-08-25)
 ....................

arq/drain.py

@@ -172,12 +172,18 @@ def _job_callback(self, task):
             self.task_exception = task_exception
         elif task.result():
             self.jobs_failed += 1
-        self.pending_tasks.remove(task)
+        self._remove_task(task)
         jobs_logger.debug('task complete, %d jobs done, %d failed', self.jobs_complete, self.jobs_failed)
 
     def _cancel_job(self, task, job):
         if not task.cancel():
             return
         self.jobs_timed_out += 1
         jobs_logger.error('task timed out %r', job)
-        self.pending_tasks.remove(task)
+        self._remove_task(task)
+
+    def _remove_task(self, task):
+        try:
+            self.pending_tasks.remove(task)
+        except KeyError:
+            pass

demo/Dockerfile

@@ -0,0 +1,21 @@
+# docker image to run the arq demo
+# image must be built using build.sh
+FROM python:3.6-alpine
+
+LABEL maintainer "[email protected]"
+
+RUN apk --update --no-cache add gcc g++ musl-dev libuv make \
+ && rm -rf /var/cache/apk/*
+
+RUN pip install -U pip setuptools
+ADD ./docker-requirements.txt /home/root/
+RUN pip install -r /home/root/docker-requirements.txt
+
+ADD ./setup.py /home/root/src/
+ADD ./arq /home/root/src/arq
+RUN pip install -e /home/root/src/
+ADD ./app.py /home/root/
+
+ENV PYTHONUNBUFFERED 1
+WORKDIR /home/root
+ENTRYPOINT ["./app.py"]

demo/README.md

@@ -0,0 +1,26 @@
+# arq demo with docker compose and aiohttp
+
+The directory contains files required to serve the demo docker/aiohttp example.
+
+## Usage
+
+**(all from the project root directory)**
+
+To build:
+
+    ./demo/build.sh
+
+To run the compose example:
+
+    export COMPOSE_FILE='demo/docker-compose.yml'
+    export COMPOSE_PROJECT_NAME='arq'
+    docker-compose up -d
+
+You'll want to then connect to logspout to view the log with something like
+
+    curl -q -s http://localhost:5001/logs
+
+You can then stop and start the worker and watch jobs get cancelled and re-enqueued
+
+    docker-compose stop worker
+    docker-compose start worker

demo/app.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3.6
+import os
+import asyncio
+from time import time
+
+import chevron
+import uvloop
+from aiohttp import web, ClientError, ClientSession
+from aiohttp_session import SimpleCookieStorage, get_session
+from aiohttp_session import setup as session_setup
+from arq import Actor, BaseWorker, RedisSettings, concurrent
+
+R_OUTPUT = 'output'
+
+asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+
+
+class Downloader(Actor):
+    re_enqueue_jobs = True
+
+    async def startup(self):
+        self.session = ClientSession(loop=self.loop)
+
+    @concurrent
+    async def download_content(self, url, count):
+        total_size = 0
+        errors = []
+        start = time()
+        for _ in range(count):
+            try:
+                async with self.session.get(url) as r:
+                    content = await r.read()
+                    total_size += len(content)
+                    if r.status != 200:
+                        errors.append(f'{r.status} length: {len(content)}')
+            except ClientError as e:
+                errors.append(f'{e.__class__.__name__}: {e}')
+        output = f'{time() - start:0.2f}s, {count} downloads, total size: {total_size}'
+        if errors:
+            output += ', errors: ' + ', '.join(errors)
+        async with self.redis_pool.get() as redis:
+            await redis.rpush(R_OUTPUT, output.encode())
+        return total_size
+
+    async def shutdown(self):
+        self.session.close()
+
+
+html_template = """
+<h1>arq demo</h1>
+
+{{#message}}
+<div>{{ message }}</div>
+{{/message}}
+
+<form method="post" action="/start-job/">
+  <p>
+  <label for="url">Url to download</label>
+  <input type="url" name="url" id="url" value="https://httpbin.org/get" required/>
+  </p>
+  <p>
+  <label for="count">Download count</label>
+  <input type="number" step="1" name="count" id="count" value="10" required/>
+  </p>
+  <p>
+  <input type="submit" value="Download"/>
+  </p>
+</form>
+
+<h2>Results:</h2>
+{{#results}}
+<p>{{ . }}</p>
+{{/results}}
+"""
+
+
+async def index(request):
+    async with await request.app['downloader'].get_redis_conn() as redis:
+        data = await redis.lrange(R_OUTPUT, 0, -1)
+        results = [r.decode() for r in data]
+
+    session = await get_session(request)
+    html = chevron.render(html_template, {'message': session.get('message'), 'results': results})
+    session.invalidate()
+    return web.Response(text=html, content_type='text/html')
+
+
+async def start_job(request):
+    data = await request.post()
+    session = await get_session(request)
+    try:
+        url = data['url']
+        count = int(data['count'])
+    except (KeyError, ValueError) as e:
+        session['message'] = f'Invalid input, {e.__class__.__name__}: {e}'
+    else:
+        await request.app['downloader'].download_content(url, count)
+        session['message'] = f'Downloading "{url}" ' + (f'{count} times.' if count > 1 else 'once.')
+    raise web.HTTPFound(location='/')
+
+
+redis_settings = RedisSettings(host=os.getenv('REDIS_HOST', 'localhost'))
+
+
+async def shutdown(app):
+    await app['downloader'].close()
+
+
+def create_app():
+    app = web.Application()
+    app.router.add_get('/', index)
+    app.router.add_post('/start-job/', start_job)
+    app['downloader'] = Downloader(redis_settings=redis_settings)
+    app.on_shutdown.append(shutdown)
+    session_setup(app, SimpleCookieStorage())
+    return app
+
+
+class Worker(BaseWorker):
+    # used by `arq app.py` command
+    shadows = [Downloader]
+    # set to small value so we can play with timeouts
+    timeout_seconds = 10
+
+    def __init__(self, *args, **kwargs):
+        kwargs['redis_settings'] = redis_settings
+        super().__init__(*args, **kwargs)
+
+
+if __name__ == '__main__':
+    # when called directly run the webserver
+    app = create_app()
+    web.run_app(app, port=8000)

demo/build.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+
+set -e
+
+THIS_DIR=`dirname "$0"`
+
+cd ${THIS_DIR}
+if [[ ! -d tmp ]]; then
+    echo "creating tmp directory..."
+    mkdir tmp
+else
+    echo "tmp directory already exists"
+fi
+
+echo "copying necessary files into place..."
+rsync -i -a requirements.txt tmp/docker-requirements.txt
+rsync -i -a --delete --exclude=*.pyc --exclude=__pycache__ ../arq tmp/
+rsync -i -a ../setup.py tmp/
+rsync -i -a app.py tmp/
+rsync -i -a Dockerfile tmp/
+
+echo "building docker image..."
+docker build tmp -t arq-demo
+echo "done."
+

demo/docker-compose.yml

@@ -0,0 +1,46 @@
+version: '3.2'
+
+volumes:
+  redis: {}
+
+services:
+  logs:
+    image: gliderlabs/logspout
+    environment:
+      SYSLOG_HOSTNAME: 'arq'
+    volumes:
+    - /var/run/docker.sock:/var/run/docker.sock
+    ports:
+    - 5001:80
+
+  redis:
+    image: redis:4.0-alpine
+    volumes:
+    - redis:/data
+    ports:
+    # redis-cli -p 63790
+    - 63790:6379
+    restart: always
+    depends_on:
+    - logs
+
+  web:
+    image: arq-demo
+    restart: always
+    ports:
+    - 8000:8000
+    environment:
+      REDIS_HOST: redis
+    depends_on:
+    - redis
+    - logs
+
+  worker:
+    image: arq-demo
+    entrypoint: ['arq', 'app.py']
+    environment:
+      REDIS_HOST: redis
+    restart: always
+    depends_on:
+    - redis
+    - logs

demo/requirements.txt

@@ -0,0 +1,4 @@
+aiohttp==2.3.2
+aiohttp-session==1.2.0
+chevron==0.11.1
+uvloop==0.8.1

docs/examples/aiohttp.py

@@ -0,0 +1,27 @@
+from aiohttp import web
+
+
+async def start_job(request):
+    data = await request.post()
+    # this will enqueue the download_content job
+    await request.app['downloader'].download_content(data['url'])
+    raise web.HTTPFound(location='/wherever/')
+
+
+async def shutdown(app):
+    await app['downloader'].close()
+
+
+def create_app():
+    app = web.Application()
+    ...
+    app.router.add_post('/start-job/', start_job)
+    app['downloader'] = Downloader()
+    # use aiohttp's on_shutdown trigger to close downloader
+    app.on_shutdown.append(shutdown)
+    return app
+
+
+if __name__ == '__main__':
+    app = create_app()
+    web.run_app(app, port=8000)

docs/usage.rst

@@ -33,6 +33,17 @@ In other words: if you need these coroutines to be called when using an actor in
 For example, in the above example there's no need for ``self.session`` when using the actor in "default" mode,
 eg. called with ``python demo.py``, so neither ``startup`` or ``shutdown`` are called.
 
+Usage with aiohttp
+..................
+
+Assuming you have ``Downloader`` already defined as per above.
+
+.. literalinclude:: examples/aiohttp.py
+
+(Won’t run as ``Downloader`` is not defined)
+
+For a full example *arq* usage with aiohttp and docker see the `demo`_ app.
+
 Health checks
 .............
 
@@ -114,3 +125,5 @@ provided ``Downloader`` and ``FooBar`` are defined and imported it should run "a
 See :meth:`arq.worker.BaseWorker` for more customisation options.
 
 For more information on logging see :meth:`arq.logs.default_log_config`.
+
+.. _demo: https://github.com/samuelcolvin/arq/tree/master/arq/demo

setup.py

@@ -3,9 +3,12 @@
 from importlib.machinery import SourceFileLoader
 from setuptools import setup
 
-with Path(__file__).resolve().parent.joinpath('README.rst').open() as f:
-    long_description = f.read()
-
+readme = Path(__file__).parent.joinpath('README.rst')
+if readme.exists():
+    with readme.open() as f:
+        long_description = f.read()
+else:
+    long_description = '-'
 # avoid loading the package before requirements are installed:
 version = SourceFileLoader('version', 'arq/version.py').load_module()