Custom exception handler in anyio

[deivydaslipskis]

Hi, I am rewriting my codebase from asyncio to anyio.

Here is my asyncio code:

import asyncio
from my_lib import TaskManager


async def process(task_manager: "TaskManager") -> None:
    loop = asyncio.get_event_loop()
    loop.set_exception_handler(task_manager.handle_exception)
    loop.create_task(task_manager.process())
    loop.run_forever()


def solve(full_name: str)) -> None:
    task_manager = TaskManager(full_name)
    return process(task_manager)

Exception handling:

    @staticmethod
    def handle_exception(loop: Any, context: Any) -> None:
        exception = context.get("exception")
        try:
            raise exception
        except TaskDone:
            logger.info("Task is completed, finishing...")
            loop.stop()
        # catching all exceptions
        except BaseException as err:
            logger.error(err, exc_info=True)

When changing to anyio I don't know how I can set my custom exception handler, how could I do that?

Here is how I currently changed my code:

import anyio
from my_lib import TaskManager

async def process(task_manager: "TaskManager") -> None:
    await task_manager.process()

async def main_proc(task_manager: "TaskManager") -> None:
    async with anyio.create_task_group() as task_group:
        task_group.start_soon(process, task_manager)

def solve(full_name: str)) -> None:
    task_manager = TaskManager(full_name)
    return process(task_manager)

I want to test anyio with different backends to test performance, are my changes the correct way to migrate from asyncio to anyio? The code seems to work but it crashes on TaskDone exception

[smurfix]

When changing to anyio I don't know how I can set my custom exception handler, how could I do that?

You don't. Structured Concurrency means that there's no need for installing a global exception handler.

async def main_proc(task_manager: "TaskManager") -> None:
    try:
        async with anyio.create_task_group() as task_group:
            task_group.start_soon(process, task_manager)
    except TaskDone:
        logger.info("Task is completed, finishing...")
    except Exception as err:  # you should never catch a BaseException without re-raising it
            logger.error(err, exc_info=True)

[deivydaslipskis]

Thank you, now it works, I also added raise err at the last line as you advised

[smurfix]

You only need the raise when you catch BaseException instead of Exception, which you shouldn't do in the first place.
(At least, not in your example.)

[deivydaslipskis]

So this would be the correct solution, right?

    except BaseException as err:
        logger.error(err, exc_info=True)
        raise err

[smurfix]

Basically yes. You can add refinements, e.g. only raise if not isinstance(err, Exception), or wrap the anyio.run call instead … but that's icing on the cake.

[deivydaslipskis]

Thanks!