Check whether an exception is a connection error

is_connection_err.py

"""
Recognize a connection error from an exception object.

Blog post:
https://gmpy.dev/blog/2023/recognize-connection-errors

Author: Giampaolo Rodola
License: MIT
"""

import errno, socket, ssl

import botocore.exceptions
import requests.exceptions

# Network errors, usually related to DHCP or wpa_supplicant (Wi-Fi).
NETWORK_ERRNOS = {
    errno.ENETUNREACH,  # "Network is unreachable"
    errno.ENETDOWN,  # "Network is down"
    errno.ENETRESET,  # "Network dropped connection on reset"
    errno.ENOTCONN,  # "Transport endpoint is not connected"
    errno.EBADF,  # "Bad file descriptor"
}
if hasattr(errno, "ENONET"):
    NETWORK_ERRNOS.add(errno.ENONET)  # "Machine is not on the network"

# requests lib connection errors
REQUESTS_EXCEPTIONS = (
    requests.exceptions.ConnectionError,
    requests.exceptions.ProxyError,
    requests.exceptions.SSLError,
    requests.exceptions.Timeout,
    requests.exceptions.ConnectTimeout,
    requests.exceptions.ReadTimeout,
    requests.exceptions.ChunkedEncodingError,
)

# botocore lib connection errors
BOTOCORE_EXCEPTIONS = (
    botocore.exceptions.ConnectionClosedError,
    botocore.exceptions.ConnectionError,
    botocore.exceptions.ConnectTimeoutError,
    botocore.exceptions.EndpointConnectionError,
    botocore.exceptions.ReadTimeoutError,
    botocore.exceptions.SSLError,
)


def is_connection_err(exc):
    """Return True if an exception is connection-related."""
    if isinstance(exc, ConnectionError):
        # https://docs.python.org/3/library/exceptions.html#ConnectionError
        # ConnectionError includes:
        # * BrokenPipeError (EPIPE, ESHUTDOWN)
        # * ConnectionAbortedError (ECONNABORTED)
        # * ConnectionRefusedError (ECONNREFUSED)
        # * ConnectionResetError (ECONNRESET)
        return True
    if isinstance(exc, socket.gaierror):
        # failed DNS resolution on connect()
        return True
    if isinstance(exc, (socket.timeout, TimeoutError)):
        # Timeout on connect(), recv(), send().
        return True
    if isinstance(exc, OSError):
        if exc.errno in NETWORK_ERRNOS:
            return True
    if isinstance(exc, ssl.SSLError):
        # Let's consider any SSL error a connection error. Usually this is:
        # * ssl.SSLZeroReturnError: "TLS/SSL connection has been closed"
        # * ssl.SSLError: [SSL: BAD_LENGTH]
        return True
    if isinstance(exc, REQUESTS_EXCEPTIONS):
        # Any indication that requests lib failed due to a connection
        # error.
        return True
    if isinstance(exc, BOTOCORE_EXCEPTIONS):
        # Any indication that boto3 lib failed due to a connection
        # error.
        return True
    return False


# =====================================================================
# --- unit tests
# =====================================================================

import unittest


class TestIsConnectionErr(unittest.TestCase):
    def test_connection_error(self):
        for exc in (
            BrokenPipeError(),
            ConnectionAbortedError(),
            ConnectionRefusedError(),
            ConnectionResetError(),
        ):
            assert is_connection_err(exc)

    def test_not_connection_error(self):
        assert not is_connection_err(ValueError())
        assert not is_connection_err(OSError())
        assert not is_connection_err(Exception())

    def test_requests_exceptions(self):
        for exc in (
            requests.exceptions.ConnectionError(),
            requests.exceptions.Timeout(),
            requests.exceptions.SSLError(),
        ):
            assert is_connection_err(exc)

    def test_botocore_exceptions(self):
        for exc in (
            botocore.exceptions.ConnectionClosedError(endpoint_url="x"),
            botocore.exceptions.ConnectTimeoutError(endpoint_url="x"),
            botocore.exceptions.ReadTimeoutError(endpoint_url="x"),
        ):
            assert is_connection_err(exc)


if __name__ == "__main__":
    unittest.main()

Hi @giampaolo, I found this piece of code very useful. Would you consider attaching any OSS license, so that people can actually use it in their projects? That would be great!

@lepus2589 Hi. Done (MIT license). I also added error handling for requests and botocore libs, which was missing + unit tests.

That's great, thanks! BTW, errno.ENONET is not available on Windows and MacOS. I fixed that with errno.ENONET if hasattr(errno, "ENONET") else 64.

Updated

mypy complains: error: "frozenset[int]" has no attribute "add"

Oh, you removed it. My bad.