Best way to schedule asynchronous HTTP requests

[BabakAmini]

I have a situation where I need to send asynchronous requests every few seconds to several URLs that are in a list. I found out that aiohttp is the best choice for my case. Below is a sample code to do this:

import asyncio
import aiohttp
import time

from codetiming import Timer


async def fetch(url, session) -> str:
    async with session.get(url) as resp:
        return await resp.text()


async def main():
    urls = [url1, url2, ...]
    update_interval = 3
    start_time = time.time()

    while True:
        tasks = []
        async with aiohttp.ClientSession() as session:
            for url in urls:
                tasks.append(asyncio.create_task(fetch(url, session)))

            resp = await asyncio.gather(*tasks, return_exceptions=True)

        time.sleep(update_interval - ((time.time() - start_time) % update_interval))


with Timer(text='Elapsed time: {:.2f}'):
    asyncio.run(main())

The above code works fine until, for whatever reason, there is a problem requesting one or more URLs. So, I thought I need a scheduler to do the job and I found AsyncScheduler useful for doing jobs asynchronously, however, I don't have any idea how I can use it to request multiple links.

I would appreciate it if one can help me in this regard.

[agronholm]

Is there a reason why you felt aiohttp was superior to httpx? I can give you a simple example involving httpx and AnyIO (which it relies on):

from __future__ import annotations

import asyncio

from anyio import create_task_group
from httpx import AsyncClient, HTTPError


async def main() -> None:
    urls = ["https://example.org/", "https://github.com/"]
    responses: list[str] = []
    update_interval = 3

    async def fetch(url: str) -> None:
        try:
            resp = await http.get(url)
        except HTTPError as exc:
            print(f"Error fetching {url}: {exc}")
        else:
            responses.append(resp.text)

    loop = asyncio.get_running_loop()
    async with AsyncClient() as http:
        while True:
            start_time = loop.time()
            async with create_task_group() as tg:
                for url in urls:
                    tg.start_soon(fetch, url)

            print("----------------\n".join(responses))
            responses.clear()
            sleep_time = max(update_interval - (loop.time() - start_time), 0)
            await asyncio.sleep(sleep_time)

asyncio.run(main())

I can make another example that uses AsyncScheduler instead if you like.

[BabakAmini]

@agronholm thanks for the reply. Regarding the best package for asynchronous requests, I did a lot of research and read many articles and I realized most of the benchmarks are for the benefit of aiohttp. I also noticed that aiohttp has better performance over others for requesting my specific URLs.

On the other hand, I tested your code for time performance by removing the while loop as well as any print statements and changed urls list to be urls = ["https://example.org/", "https://github.com/"] * 20. I also changed aiohttp version in the same way. Both methods take the same time to finish while aiohttp version is more readable and less complicated.

These made me choose aiohttp.

Finally, I'm interested to see an example of using AsyncScheduler. I'm also wondering if the performance of that example is the same as the above methods. Please note that if one or more requests fail for any reason and the time exceeds update_interval, then none of the above examples works as expected. This is the reason I'm willing to use AsyncScheduler so that all the situations are handled by your package.

[agronholm]

My code had differences that don't really have anything to do with the differences between httpx vs asyncio.
In particular:

1. It uses a single session instead of (unnecessarily) creates multiple sessions
2. It handles exceptions occurring during the HTTP queries
3. It prints the results on the console

Regarding the features of httpx vs asyncio, I'd like to point out the following:

1. httpx supports HTTP/2, aiohttp does not
2. httpx also has a synchronous interface

Here is the equivalent code using AsyncScheduler:

from __future__ import annotations

import asyncio

from anyio import create_task_group
from httpx import AsyncClient, HTTPError

from apscheduler.schedulers.async_ import AsyncScheduler
from apscheduler.triggers.interval import IntervalTrigger


async def fetch_urls(urls: list[str], http: AsyncClient) -> None:
    responses: list[str] = []

    async def fetch(url: str) -> None:
        try:
            resp = await http.get(url)
        except HTTPError as exc:
            print(f"Error fetching {url}: {exc}")
        else:
            responses.append(resp.text)

    async with create_task_group() as tg:
        for url in urls:
            tg.start_soon(fetch, url)

    print("----------------\n".join(responses))


async def main() -> None:
    urls = ["https://example.org/", "https://github.com/"]

    async with AsyncClient() as http, AsyncScheduler() as scheduler:
        await scheduler.add_schedule(
            fetch_urls, IntervalTrigger(seconds=2), args=[urls, http]
        )
        await scheduler.wait_until_stopped()


asyncio.run(main())

[agronholm]

For comparison, here is the same, but with aiohttp:

from __future__ import annotations

import asyncio

from aiohttp import ClientSession, ClientError
from anyio import create_task_group

from apscheduler.schedulers.async_ import AsyncScheduler
from apscheduler.triggers.interval import IntervalTrigger


async def fetch_urls(urls: list[str], http: ClientSession) -> None:
    responses: list[str] = []

    async def fetch(url: str) -> None:
        try:
            async with http.get(url) as resp:
                responses.append(await resp.text())
        except ClientError as exc:
            print(f"Error fetching {url}: {exc}")

    async with create_task_group() as tg:
        for url in urls:
            tg.start_soon(fetch, url)

    print("----------------\n".join(responses))


async def main() -> None:
    urls = ["https://example.org/", "https://github.com/"]

    async with ClientSession() as http, AsyncScheduler() as scheduler:
        await scheduler.add_schedule(
            fetch_urls, IntervalTrigger(seconds=2), args=[urls, http]
        )
        await scheduler.wait_until_stopped()


asyncio.run(main())

[BabakAmini]

Since the server I'm requesting doesn't support http2, this is not critical for me, but that was a good point about multiple sessions; I fully agree with you, so it's better to move async with outside the while loop.

Your sample code for AsyncScheduler is just what I was looking for, and it's working like a charm. Thanks for your help.

[BabakAmini]

Since there is no option to make replies as accepted answers, I referenced your good points here again.

[Herojos]

Hi, i'm tryining to implement this code for my purpose. I've some issue on his line: from apscheduler.schedulers.async_ import AsyncScheduler. My terminal shows: apscheduler.schedulers.async_ could not be resolved. Can I resolve it?

[agronholm]

Are you sure you're using v4.0 alpha?

[Herojos]

You are right. My pip list shows this package: APScheduler 3.9.1.post1. Can you help me to install the corret version? How to?
Tnk for help, i'm new.

[agronholm]

pip install apscheduler==4.0.0a2