all_settled.py

from __future__ import annotations
import asyncio
from typing import List, Any, Coroutine, TypeVar, Generic, TypeAlias
from dataclasses import dataclass

T = TypeVar('T')

@dataclass
class Fulfilled(Generic[T]):
    value: T

@dataclass
class Rejected:
    error: BaseException

SettledResult: TypeAlias = Fulfilled[T] | Rejected

async def all_settled(coroutines: List[Coroutine[Any, Any, T]]) -> List[SettledResult[T]]:
    async def handle_coroutine(coroutine: Coroutine[Any, Any, T]) -> SettledResult[T]:
        try:
            result = await coroutine
            return Fulfilled(result)
        except BaseException as e:
            return Rejected(e)

    return await asyncio.gather(*(handle_coroutine(coro) for coro in coroutines))

# Example usage
async def success() -> str:
    await asyncio.sleep(1)
    return "Success"

async def failure() -> None:
    await asyncio.sleep(1)
    raise ValueError("Failure")

async def main() -> None:
    results = await all_settled([success(), failure()])
    
    for result in results:
        match result:
            case Fulfilled(value):
                print(f"Fulfilled with value: {value}")
            case Rejected(error):
                print(f"Rejected with error: {type(error).__name__}: {error}")
                print("Traceback:")
                import traceback
                traceback.print_exception(type(error), error, error.__traceback__)

if __name__ == "__main__":
    asyncio.run(main())