producer_consumer.py

gistfile1.txt

#! /usr/bin/env python3

import asyncio
from concurrent.futures import ThreadPoolExecutor
import requests
import sys


def background_task():
    rv = requests.get('https://httpbin.org/get')
    data = rv.json()
    return data


async def produce(queue, pool):
    loop = asyncio.get_running_loop()
    for i in range(5):
        t = loop.run_in_executor(pool, background_task)
        await queue.put(t)
        await asyncio.sleep(1)
    await queue.put(None)


async def consume(queue):
    while True:
        t = await queue.get()
        try:
            if not t:
                break
            data = await t
            print('consumed', data)
        finally:
            queue.task_done()

async def main():
    q = asyncio.Queue()
    with ThreadPoolExecutor() as pool:
        await asyncio.wait([produce(q, pool), consume(q)])
        await q.join()
    return 0


sys.exit(asyncio.run(main()))