Making Parallel HTTP Requests With aiohttp

Hi,

I have a couple of questions:

1) What is the state of the python thread when we are ‘awaiting’ on I/O processing? Is the python thread technically sleeping and woken up later by the operating system or is it just waiting/polling? 2) Why do we require the await keyword while attempting to get the text using response.text()? Is it because response becomes a coroutine object rather than a normal python object?

Guys, any updates?

Where exactly the http GET request gets executed ? Is it in the response = await session.request(method='GET', url=url)statement or in the next statement i.e., value = response.text() statement ? If the request happens in only one of these statements, then why do we need await for both the statements ?

Hi there Chyld. Great session, really liked it.

I was just wondering if there was a way to define a generator function like we did in the theory session and use that for the async http call? As opposed to using one line generator expression?

Say that the generator function would look like this:

   def gen_workers(stop):
        for member in range(1, stop + 1):
        yield member, randint(1, 10)

I played around with it but couldn’t figure out how to make the call asynchronous. The below works synchronously, but not in async!

            members = list(gen_workers(1))
            for member in members:
                responses = await asyncio.gather(worker(member[0], member[1], session))
            print(responses)

Thanks again /Par

Ah, never mind. I read the tutorial of concurrency and figured out that we can use a different type of syntax for iterating over an already generated list.

Generator function stays the same:

    def gen_workers(stop):
        for member in range(1, stop + 1):
        yield member, randint(1, 10)

We do this for the async function:

    async def alternate_worker_pool():
        async with aiohttp.ClientSession() as session:
            # Synchronous way with external generator
            members = list(gen_workers(10))
            tasks = []
            for member in members:
                task = asyncio.ensure_future(worker(member[0], member[1], session))
                tasks.append(task)
            await asyncio.gather(*tasks, return_exceptions=True)

And we run the event loop with:

    if __name__ == '__main__':
        start = time.perf_counter()
        asyncio.get_event_loop().run_until_complete(
             alternate_worker_pool())
        elapsed = time.perf_counter() - start
        print(f'executed in {elapsed:0.2f} seconds')

Thanks for a great video. Cheers

So the speed up is from hoping that the website you formed the request to will parallelize the task on their machine? Isn’t Python running the tasks on different cores asynchronously?

The questions above are somewhat recent and deserving of an answer. Requests for I/O are in the domain of the operating system. Those requests are therefore conducted in that process, outside of the process where your python code is being executed. As such, one would need to review the O/S API. It has been some time since the last time I reviewed an O/S API but at that time I recall a set of calls that the calling process would provide a callback to notify when an operation completed and one that return a result when the operation completed.

The first call was non-blocking and the second, blocking.

One might consider that the await is setting up a callback for the O/S to call when a specific operation completes. O/S API calls follow a general pattern as well. Often a process will make a request for data without knowing how much data will be returned. It is a simple matter to know how large a file is in a file system but other data sources can be unbounded streams of data. In the latter cases, a program has to allocate memory to store incoming data and provide a pointer to where that memory is and a counter of how much memory is in the buffer (a counter to provide a count of how many bytes are in the buffer) to the O/S. The O/S then places the incoming data into the buffer and returns a count of how many bytes were required. If the amount of data being retrieved is larger than the buffer, which is most the time, then this is an iterative process. Hence the second await in the second line of code. When python passes a buffer off to the O/S for the O/S to fill it, python also provides a callback for the O/S to call to notify python when the operation is complete. The buffer fills in the O/S thread leaving python free to perform other tasks in that time frame.

When the O/S makes the callback, it is a matter of implementation as to if the callback actually performs work or sets a flag for the event loop to attend to the task during normal processing. In any case, the callback should complete processing as rapidly as possible. For this reason and based on the syntax flow of python in this case, it seems reasonable that the latter is more likely the case.

The async and await key words are extensions to the python language. They facilitate adding housekeeping tasks to an event loop. Conceptually, async tells python that a particular method requires a callback framework and await indicates where those callbacks are implemented within the method. Each await, when encountered, is an opportunity for control to be passed to a different task via the event loop.

When I try to run the summer.py code using asyncio.run(main()), I am getting a RunTimeError saying:

Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "C:\ProgramData\Anaconda3\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\ProgramData\Anaconda3\lib\asyncio\base_events.py", line 746, in call_soon
    self._check_closed()
  File "C:\ProgramData\Anaconda3\lib\asyncio\base_events.py", line 510, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

Was looking on stackexchange, it seems that running asyncio.get_event_loop().run_until_complete(main()) does not return this error. Any insight into why this is? Thanks.

With summer.py and range(2,30) I get an error:

Traceback (most recent call last):
  File "/Volumes/Work/rp_examples/asynchrony/summer.py", line 35, in <module>
    asyncio.run(main())
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/Volumes/Work/rp_examples/asynchrony/summer.py", line 30, in main
    sums = await asyncio.gather(*(worker(f"w{i}", n, session) for i, n in enumerate(range(2, 20))))
  File "/Volumes/Work/rp_examples/asynchrony/summer.py", line 18, in worker
    raise e
  File "/Volumes/Work/rp_examples/asynchrony/summer.py", line 15, in worker
    value = json.loads(value)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/json/__init__.py", line 346, in loads
    return _default_decoder.decode(s)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Because of this:

Problem with value='The QRNG API is limited to 1 requests per second. 
For more requests, please visit https://quantumnumbers.anu.edu.au 
or contact qrng@anu.edu.au.'

Now this website only accepts 1 request per minute ‘The QRNG API is limited to 1 requests per minute. For more requests, please visit quantumnumbers.anu.edu.au or contact qrng@anu.edu.au.’ I used a different url to get a similar behaviour(time consumption is different):

async def worker(name, n, session):
    print(f'worker-{name}')
    # url = f'http://qrng.anu.edu.au/API/jsonI.php?length={n}&type=uint16'
    url = f'https://www.random.org/integers/?num={n}&min=1&max=100&col=1&base=10&format=html&rnd=new'
    response = await session.request(method='GET', url=url)
    value = await response.text()
    match = re.search(r'<pre class="data">([\d\s\n]+)</pre>', value)
    if match:
        data_content = match.group(1)  # Extract the random numbers
        # Clean up the data
        random_numbers = [int(num) for num in data_content.split() if num.strip()]
        print(random_numbers)
        value = {'data': random_numbers}
    return value['data']

@aromagosa

You can Use this API instead, it accept multiple requests at same time.

API documentation: www.randomnumberapi.com/

API endpoint: www.randomnumberapi.com/api/v1.0/random?min=100&max=1000&count=5