Python 3.11 Exception Groups

exception_groups.py

from urllib.request import urlopen
from urllib.error import HTTPError
from concurrent.futures import ThreadPoolExecutor, as_completed


class HttpError(Exception):

    def __init__(self, url: str, code: int) -> None:
        self.url = url
        self.code = code


class HttpStatusError(HttpError):
    pass


class ServerError(HttpError):
    pass


def get_status(url):
    try:
        response = urlopen(url)
    except HTTPError as e:
        if 400 <= e.code < 500:
            raise HttpStatusError(url, e.code)
        elif e.code >= 500:
            raise ServerError(url, e.code)
    else:
        return response.status


def execute():
    urls = [
        'https://khashtamov.com',
        'https://khashtamov.com/status/404/',
        'https://httpbin.org/status/500',
        'https://httpbin.org/status/502',
    ]
    exceptions = []
    with ThreadPoolExecutor(max_workers=len(urls)) as executor:
        futures = {
            executor.submit(
                get_status,
                url,
            ): url
            for url in urls
        }
        for future in as_completed(futures):
            try:
                status_code = future.result()
            except Exception as e:
                exceptions.append(e)
            else:
                print(f'{futures[future]} = {status_code}')

        if exceptions:
            raise BaseExceptionGroup('Unhandled exceptions', exceptions)


if __name__ == '__main__':
    try:
        execute()
    except* ServerError as e:
        print('Server error occurred')
        for exception in e.exceptions:
            print(f'{exception.url} = {exception.code}')

    except* HttpStatusError as e:
        print('Http error occurred')
        for exception in e.exceptions:
            print(f'{exception.url} = {exception.code}')