Examples using asyncio

Waiting in asyncio has a very good overview about which wait function to use when and explains all differences.

Mastodon

Using the Mastodon API with asyncio and httpx.

Example 1 - Mastodon

import asyncio
import json
import sys

import httpx

SERVERS = [
    "https://mastodon.social",
    "https://norden.social",
    "https://fosstodon.org",
    "https://floss.social",
    "https://osna.social",
]


async def get_public_feed(
    client: httpx.AsyncClient, server: str
) -> tuple[str, dict]:
    response = await client.get(f"{server}/api/v1/timelines/public")
    response.raise_for_status()
    return server, response.json()


async def get_trends(
    client: httpx.AsyncClient, server: str
) -> tuple[str, dict]:
    response = await client.get(f"{server}/api/v1/timelines/public")
    response.raise_for_status()
    return server, response.json()


async def main():
    servers = sys.argv[1:] if len(sys.argv) > 1 else SERVERS

    async with httpx.AsyncClient(
        http2=True, timeout=httpx.Timeout(20.0, connect=30.0)
    ) as client:

        trends_coros = [get_trends(client, server) for server in servers]
        for coro in asyncio.as_completed(trends_coros):
            server, data = await coro
            print(server, json.dumps(data, indent=2))

        # asyncio.wait expects tasks
        # passing coroutines prints a deprecation warning
        public_feed_tasks = [
            asyncio.create_task(get_public_feed(client, server))
            for server in servers
        ]
        done, pending = await asyncio.wait(public_feed_tasks, timeout=0.5)
        print(f"{len(done)} requests done, {len(pending)} request pending")
        for task in done:
            server, data = await task
            print(server, json.dumps(data, indent=2))

        if pending:
            done, pending = await asyncio.wait(pending)


asyncio.run(main())