Source code for aiohttp.client_exceptions

"""HTTP related errors."""

import asyncio


try:
    import ssl
except ImportError:  # pragma: no cover
    ssl = None


__all__ = (
    'ClientError',

    'ClientConnectionError',
    'ClientOSError', 'ClientConnectorError', 'ClientProxyConnectionError',

    'ClientSSLError',
    'ClientConnectorSSLError', 'ClientConnectorCertificateError',

    'ServerConnectionError', 'ServerTimeoutError', 'ServerDisconnectedError',
    'ServerFingerprintMismatch',

    'ClientResponseError', 'ClientHttpProxyError',
    'WSServerHandshakeError', 'ContentTypeError',

    'ClientPayloadError', 'InvalidURL')


[docs]class ClientError(Exception): """Base class for client connection errors."""
[docs]class ClientResponseError(ClientError): """Connection error during reading response. request_info: instance of RequestInfo """ def __init__(self, request_info, history, *, code=0, message='', headers=None): self.request_info = request_info self.code = code self.message = message self.headers = headers self.history = history super().__init__("%s, message='%s'" % (code, message))
[docs]class ContentTypeError(ClientResponseError): """ContentType found is not valid."""
[docs]class WSServerHandshakeError(ClientResponseError): """websocket server handshake error."""
class ClientHttpProxyError(ClientResponseError): """HTTP proxy error. Raised in :class:`aiohttp.connector.TCPConnector` if proxy responds with status other than ``200 OK`` on ``CONNECT`` request. """
[docs]class ClientConnectionError(ClientError): """Base class for client socket errors."""
[docs]class ClientOSError(ClientConnectionError, OSError): """OSError error."""
[docs]class ClientConnectorError(ClientOSError): """Client connector error. Raised in :class:`aiohttp.connector.TCPConnector` if connection to proxy can not be established. """ def __init__(self, connection_key, os_error): self._conn_key = connection_key self._os_error = os_error super().__init__(os_error.errno, os_error.strerror) @property def os_error(self): return self._os_error @property def host(self): return self._conn_key.host @property def port(self): return self._conn_key.port @property def ssl(self): return self._conn_key.ssl def __str__(self): return ('Cannot connect to host {0.host}:{0.port} ssl:{0.ssl} [{1}]' .format(self, self.strerror))
[docs]class ClientProxyConnectionError(ClientConnectorError): """Proxy connection error. Raised in :class:`aiohttp.connector.TCPConnector` if connection to proxy can not be established. """
[docs]class ServerConnectionError(ClientConnectionError): """Server connection errors."""
[docs]class ServerDisconnectedError(ServerConnectionError): """Server disconnected.""" def __init__(self, message=None): self.message = message
[docs]class ServerTimeoutError(ServerConnectionError, asyncio.TimeoutError): """Server timeout error."""
[docs]class ServerFingerprintMismatch(ServerConnectionError): """SSL certificate does not match expected fingerprint.""" def __init__(self, expected, got, host, port): self.expected = expected self.got = got self.host = host self.port = port def __repr__(self): return '<{} expected={} got={} host={} port={}>'.format( self.__class__.__name__, self.expected, self.got, self.host, self.port)
[docs]class ClientPayloadError(ClientError): """Response payload error."""
[docs]class InvalidURL(ClientError, ValueError): """Invalid URL. URL used for fetching is malformed, e.g. it doesn't contains host part.""" # Derive from ValueError for backward compatibility def __init__(self, url): super().__init__(url) @property def url(self): return self.args[0] def __repr__(self): return '<{} {}>'.format(self.__class__.__name__, self.url)
[docs]class ClientSSLError(ClientConnectorError): """Base error for ssl.*Errors."""
if ssl is not None: certificate_errors = (ssl.CertificateError,) certificate_errors_bases = (ClientSSLError, ssl.CertificateError,) ssl_errors = (ssl.SSLError,) ssl_error_bases = (ClientSSLError, ssl.SSLError) else: # pragma: no cover certificate_errors = tuple() certificate_errors_bases = (ClientSSLError, ValueError,) ssl_errors = tuple() ssl_error_bases = (ClientSSLError,)
[docs]class ClientConnectorSSLError(*ssl_error_bases): """Response ssl error."""
[docs]class ClientConnectorCertificateError(*certificate_errors_bases): """Response certificate error.""" def __init__(self, connection_key, certificate_error): self._conn_key = connection_key self._certificate_error = certificate_error @property def certificate_error(self): return self._certificate_error @property def host(self): return self._conn_key.host @property def port(self): return self._conn_key.port @property def ssl(self): return self._conn_key.ssl def __str__(self): return ('Cannot connect to host {0.host}:{0.port} ssl:{0.ssl} ' '[{0.certificate_error.__class__.__name__}: ' '{0.certificate_error.args}]'.format(self))