Handling of exceptions and documentation

[fersarr]

Hi! Thank you for writing this useful library!

I couldn't find any documentation about exceptions. Can we use aiometer.run_all and make it continue even if there are exceptions so that we can gather them all at the end?

Thank you.

[florimondmanca]

Hi @fersarr!

Have you already experimented what the behavior looks like?

From my memory, I don’t think I actually specified what the exceptions behavior should be. But I assume it should match what anyio does: either raise an exception from any function if it occurs, or raise an exception group if exceptions occur in multiple concurrent functions?

[fersarr]

Thanks for the quick response!

Yea, aiometer.run_all right now stops if there is an exception. I was looking for perhaps a run_all parameter that allows me to do something like return_exceptions from asyncio:

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

So this allows me to finish all the processing for all the independent concurrent tasks and then handle exceptions if I want to. Is that clear? Let me know if you need more information.

Example code which attempts 10 requests but it stops processing after there is one exception (unlike the return_exceptions behaviour):

import asyncio
import functools
import random
import aiometer
import httpx

client = httpx.AsyncClient()


async def fetch(client, request, idx):
    print(f"fetching {idx}")

    if idx == 2:
        raise Exception(f"failing at {idx}")

    response = await client.send(request)
    # Simulate extra processing...
    await asyncio.sleep(2 * random.random())
    print(f"done fetching {idx}")
    return response.json()["json"]


async def mymain():

    requests = [
        httpx.Request("POST", "https://httpbin.org/anything", json={"index": index})
        for index in range(10)
    ]

    # Alternatively, fetch and aggregate responses into an (ordered) list...
    jobs = [functools.partial(fetch, client, request, idx) for idx, request in enumerate(requests)]
    results = await aiometer.run_all(jobs, max_at_once=10, max_per_second=5)


asyncio.run(mymain())

Results in:

$ python scripts/aiometer_test.py
fetching 0
fetching 1
fetching 2
Traceback (most recent call last):
  File "/.../aiometer_test.py", line 35, in <module>
    asyncio.run(mymain())

[florimondmanca]

I think I'd instead do it this way…

import asyncio
import functools
import random
import aiometer
import httpx

client = httpx.AsyncClient()


async def fetch(client, request, idx):
    print(f"fetching {idx}")

    if idx == 2:
        raise Exception(f"failing at {idx}")

    response = await client.send(request)
    # Simulate extra processing...
    await asyncio.sleep(2 * random.random())
    print(f"done fetching {idx}")
    return response.json()["json"]


async def mymain():
    requests = [
        httpx.Request("POST", "https://httpbin.org/anything", json={"index": index})
        for index in range(10)
    ]

    exceptions = {}

    async def fetch_collecting_exceptions(client, request, idx):
        try:
            return await fetch(client, request, idx)
        except Exception as exc:
            exceptions[idx] = exc
            return None

    jobs = [
        functools.partial(fetch_collecting_exceptions, client, request, idx)
        for idx, request in enumerate(requests)
    ]

    results = await aiometer.run_all(jobs, max_at_once=10, max_per_second=5)

    # Process `results` and `exceptions`...


asyncio.run(mymain())

[fersarr]

Yea, thanks, that's a good workaround, I could always add wrappers that catch the exceptions. I guess it's just a matter of convenience of just getting that wrapper automatically with a flag like return_exceptions. Have a good one!

[florimondmanca]

Answered via #33 (reply in thread)