async_download_publish.py

#!/usr/bin/env python

"""
Playground for mocking asynchronous downloads and later publishing some stuff.
"""

import asyncio
import time
from functools import partial


SIMULTANEOUS_DOWNLOADS = 3


def timeit(f):
    def _wrapper(*args, **kwargs):
        time1 = time.time()
        result = f(*args, **kwargs)
        print("Was running for %.04f sec" % (time.time() - time1))
        return result
    return _wrapper


class SyncUrls(list):
    def __len__(self):
        return 3

    def __iter__(self):
        yield self.process("Z", 3)
        yield self.process("X", 1)
        yield self.process("Y", 2)

    def download_url(self, code, timeout):
        print("{} downloading".format(code))
        time.sleep(timeout)
        print("{} finished downloading".format(code))
        return code * 3

    def publish_url(self, code, timeout=0):
        print("{} publishing".format(code))
        if timeout:
            time.sleep(timeout)
        print("{} finished publishing".format(code))

    def process(self, code, timeout=0):
        r = self.download_url(code, timeout)
        self.publish_url(r, timeout)


class Urls(list):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.sem = asyncio.Semaphore(SIMULTANEOUS_DOWNLOADS)

    def __len__(self):
        return 3

    def __iter__(self):
        yield self.process("Z", 3)
        yield self.process("X", 1)
        yield self.process("Y", 2)

    @asyncio.coroutine
    def download_url(self, code, timeout):
        print("{} downloading".format(code))
        yield from asyncio.sleep(timeout)
        print("{} finished downloading".format(code))
        return code * 3

    def download_url_blocking(self, code, timeout):
        print("{} blocking downloading".format(code))
        time.sleep(timeout)
        print("{} finished blocking downloading".format(code))
        return code * 3

    @asyncio.coroutine
    def publish_url(self, code, timeout=0):
        print("{} publishing".format(code))
        if timeout:
            yield from asyncio.sleep(timeout)
        print("{} finished publishing".format(code))

    @asyncio.coroutine
    def process(self, code, timeout=0):
        with (yield from self.sem):
            downloader = partial(self.download_url_blocking, code, timeout)
            r = yield from asyncio.get_event_loop().run_in_executor(None, downloader)
            # r = yield from self.download_url(code, timeout)
        yield from self.publish_url(r, timeout)


@asyncio.coroutine
def download_all():
    yield from asyncio.wait(Urls())


@timeit
def download_all_sync():
    for x in SyncUrls():
        pass


@timeit
def download_all_async():
    io_loop = asyncio.get_event_loop()
    io_loop.run_until_complete(
        download_all()
    )


print("\n", "-" * 10)
print("Running SYNC")
download_all_sync()


print("\n", "-" * 10)
print("Running ASYNC")
download_all_async()