async / await example

async_producer_consumer.py

#!/usr/bin/env python3.5

import abc
import asyncio

import string
import random


class WordProducer(metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def get_word(self):
        pass

    @abc.abstractmethod
    async def get_words(self):
        pass


class WordProducer:

    def __init__(self, *args, **kwargs):
        super().__init__()

    async def get_words(self):
        return [self.get_word(), ]


class DelayedWordProducer:

    def __init__(self, prefix='', min_wait=0.1, max_wait=1.0):
        self.prefix = prefix
        self.min_wait = min_wait
        self.max_wait = max_wait

        super().__init__()

    async def get_delayed_word(self):
        await asyncio.sleep(random.uniform(self.min_wait, self.max_wait))
        return '{}{}'.format(self.prefix, self.get_word())

    async def get_words(self):
        words = []

        async def get_word(i):
            word = await self.get_delayed_word()
            words.append(word)

        word_tasks = [asyncio.ensure_future(get_word(i)) for i in range(5)]
        await asyncio.wait(word_tasks)

        return words


class RandomWordProducerMixin:

    def get_word(self, nmin=5, nmax=10):
        times = random.randint(nmin, nmax)
        return ''.join(random.choice(string.ascii_letters) for n in range(times))


class DictionaryWordProducerMixin:

    def __init__(self):
        self.words = ('house', 'tree', 'sky', 'mountain', )

    def get_word(self):
        return random.choice(self.words)


class RandomWordProducer(WordProducer, RandomWordProducerMixin):
    pass


class DictionaryWordProducer(WordProducer, DictionaryWordProducerMixin):
    pass


class DelayedDictionaryWordProducer(DelayedWordProducer, DictionaryWordProducerMixin):
    pass

class DelayedRandomWordProducer(DelayedWordProducer, RandomWordProducerMixin):
    pass


class WordConsumer:

    def __init__(self):
        self.producers = [
            RandomWordProducer(),
            DictionaryWordProducer(),
            DelayedDictionaryWordProducer('A_', 1, 5),
            DelayedRandomWordProducer('B_', 1, 5),
        ]

    def collect_words(self):
        words = []

        async def collect_words(origin):
            origin_words = await origin.get_words()
            for word in origin_words:
                words.append(word)

        production_tasks = []
        for producer in self.producers:
            production_tasks.append(collect_words(producer))

        loop = asyncio.get_event_loop()
        loop.run_until_complete(asyncio.wait(production_tasks))
        loop.close()

        return words


if __name__ == '__main__':
    consumer = WordConsumer()
    print(consumer.collect_words())