Example of asynchronous requests with a parameterizable callback function and processed item tracking, using aiohttp and asyncio.

bulk_aiohttp_callback.py

#!/usr/bin/env python3.7

import asyncio
from asyncio import Task

from aiohttp import ClientResponseError, ClientSession


# TODO Concrete example to be generalized.

TOKEN = "..."

HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}


async def _retrieve(index, id, session):
    encoded_id = f"https%3A%2F%2Fbbp.epfl.ch%2Fneurosciencegraph%2Fdata%2F{id}"
    url = f"https://bbp.epfl.ch/nexus/v1/resources/bbp/neurocurator/_/{encoded_id}"
    try:
        async with session.get(url) as response:
            _json = await response.json()
            return index, "SUCCESS", _json
    except ClientResponseError as e:
        # FIXME Find how to get the same as requests.exceptions.HTTPError.response.json().
        return index, "FAILURE", e.message


def _build_retrieve_task(index, id, session, callback):
    task = Task(_retrieve(index, id, session))
    task.add_done_callback(callback)
    return task


async def _bulk_retrieve(ids, callback):
    async with ClientSession(headers=HEADERS, raise_for_status=True) as session:
        tasks = (_build_retrieve_task(i, x, session, callback) for i, x in enumerate(ids))
        await asyncio.gather(*tasks)


def bulk_retrieve(ids, callback):
    asyncio.run(_bulk_retrieve(ids, callback))


def fcallback(task):
    print("###", *task.result(), "###", sep="\n")


bulk_retrieve(["eccf13c0-a1ae-11e6-98a4-64006a4c56ef", "invalid"], fcallback)