Python try except timeout error

In this article, we will guide you through how to set timeout in requests, as well as managing exceptions.

Requests is the de-facto standard when it comes to making HTTP requests in Python, especially since Python 3. The open-source library abstracts away the complexity of managing network connections and make sending HTTP requests a breeze.

Most of you may already know how to send HTTP POST, GET, as well as other type of HTTP requests. In this article, we will guide you through how to set timeout in requests, as well as managing exceptions.

Timeouts in Python requests

You can tell requests library to stop waiting for a response after a given amount of time by passing a number to the timeout parameter. If the requests library does not receive response in x seconds, it will raise a Timeout error.

It’s a best practice that production code should use this parameter in all network requests. Failure to do so can cause your program to hang indefinitely. If no timeout is specified explicitly, requests do not time out.

requests.get('https://example.com/', timeout=10)
# OUTPUT
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
requests.exceptions.Timeout: HTTPConnectionPool(host='example.com', port=80): Request timed out. (timeout=10)
Code language: PHP (php)

Catch timeout exception

Once a timeout value has been set, every request that doesn’t receive a response in the specified timeframe will raise a Timeout error. It’s important to handle this exception, otherwise your program would be terminated.

In order to catch Timeout errors in requests, you have to import the exception itself using from requests.exceptions import Timeout.

import requests

from requests.exceptions import Timeout
try:
    requests.get('https://www.example.com, timeout=10)
except Timeout:
    print('Timeout has been raised.')
Code language: PHP (php)

Advanced timeout handling

Most requests to external servers should have a timeout attached, in case the server is not responding in a timely manner. By default, requests do not time out unless a timeout value is set explicitly. Without a timeout, your code may hang for minutes or more.

The connect timeout is the number of seconds Requests will wait for your client to establish a connection to a remote machine (corresponding to the connect()) call on the socket. It’s a good practice to set connect timeouts to slightly larger than a multiple of 3, which is the default TCP packet retransmission window.

Once your client has connected to the server and sent the HTTP request, the read timeout is the number of seconds the client will wait for the server to send a response. (Specifically, it’s the number of seconds that the client will wait between bytes sent from the server. In 99.9% of cases, this is the time before the server sends the first byte).

If you specify a single value for the timeout, like this:

r = requests.get('https://github.com', timeout=5)
Code language: JavaScript (javascript)

The timeout value will be applied to both the connect and the read timeouts. Specify a tuple if you would like to set the values separately:

r = requests.get('https://github.com', timeout=(3.05, 27))
Code language: JavaScript (javascript)

Request timeout in async coroutines

Since there are a lot of beginners asked how to use requests in asynchronous programs or coroutines, we’ve made this section dedicated to answering just that.

requests is a blocking library, which means it should not be used in asynchronous coroutines. The proper libraries to use in this context is aiohttp and async-timeout. aiohttp is an asynchronous HTTP client/server framework, while async-timeout is a asyncio-compatible timeout context manager that can be used to wrap any coroutines inside a “timeout” context manager.

Источник

The following are 30
code examples of requests.exceptions.Timeout().
You can vote up the ones you like or vote down the ones you don’t like,
and go to the original project or source file by following the links above each example.
You may also want to check out all available functions/classes of the module
requests.exceptions
, or try the search function

.

Example #1

def request(self, url, *args, **kwargs):
        try:
            stream = HttpStream(url, *args, verbose=self.verbose, **kwargs)
            if kwargs.get('stream', False):
                return stream

            with stream as s:
                content = s.req.content
                return HttpResponse(s.status_code, s.headers, content)
        # Timeout will catch both ConnectTimout and ReadTimeout
        except (RetryError, Timeout) as ex:
            raise OsbsNetworkException(url, str(ex), '',
                                       cause=ex, traceback=sys.exc_info()[2])
        except HTTPError as ex:
            raise OsbsNetworkException(url, str(ex), ex.response.status_code,
                                       cause=ex, traceback=sys.exc_info()[2])
        except Exception as ex:
            raise OsbsException(cause=ex, traceback=sys.exc_info()[2])

Example #2

def test_do_api_call_waits_between_retries(self, mock_sleep):
        retry_delay = 5
        self.hook = DatabricksHook(retry_delay=retry_delay)

        for exception in [requests_exceptions.ConnectionError,
                          requests_exceptions.SSLError,
                          requests_exceptions.Timeout,
                          requests_exceptions.ConnectTimeout,
                          requests_exceptions.HTTPError]:
            with mock.patch('airflow.providers.databricks.hooks.databricks.requests') as mock_requests:
                with mock.patch.object(self.hook.log, 'error'):
                    mock_sleep.reset_mock()
                    setup_mock_requests(mock_requests, exception)

                    with self.assertRaises(AirflowException):
                        self.hook._do_api_call(SUBMIT_RUN_ENDPOINT, {})

                    self.assertEqual(len(mock_sleep.mock_calls), self.hook.retry_limit - 1)
                    calls = [
                        mock.call(retry_delay),
                        mock.call(retry_delay)
                    ]
                    mock_sleep.assert_has_calls(calls)

Example #3

def test_neutron_exception_is_raised_on_any_request_error(self):
        # timeout exception raises InfobloxTimeoutError
        f = mock.Mock()
        f.__name__ = 'mock'
        f.side_effect = req_exc.Timeout
        self.assertRaises(exceptions.InfobloxTimeoutError,
                          connector.reraise_neutron_exception(f))

        # all other request exception raises InfobloxConnectionError
        supported_exceptions = [req_exc.HTTPError,
                                req_exc.ConnectionError,
                                req_exc.ProxyError,
                                req_exc.SSLError,
                                req_exc.TooManyRedirects,
                                req_exc.InvalidURL]

        for ex in supported_exceptions:
            f.side_effect = ex
            self.assertRaises(exceptions.InfobloxConnectionError,
                              connector.reraise_neutron_exception(f))

Example #4

def get_pod_relay_preferences(self, host):
        """Query remote pods on https first, fall back to http."""
        logging.info("Querying %s" % host)
        try:
            try:
                response = requests.get("https://%s/.well-known/x-social-relay" % host,
                                timeout=5,
                                headers={"User-Agent": config.USER_AGENT})
            except timeout:
                response = None
            if not response or response.status_code != 200:
                response = requests.get("http://%s/.well-known/x-social-relay" % host,
                                timeout=5,
                                headers={"User-Agent": config.USER_AGENT})
                if response.status_code != 200:
                    return None
        except (ConnectionError, Timeout, timeout):
            return None
        try:
            # Make sure we have a valid x-social-relay doc
            validate(response.json(), self.schema)
            return response.text
        except (ValueError, ValidationError):
            return None

Example #5

def requests_get(url, **kwargs):
    USER_AGENTS = (
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:11.0) Gecko/20100101 Firefox/11.0',
        'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:22.0) Gecko/20100 101 Firefox/22.0',
        'Mozilla/5.0 (Windows NT 6.1; rv:11.0) Gecko/20100101 Firefox/11.0',
        ('Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_4) AppleWebKit/536.5 (KHTML, like Gecko) '
         'Chrome/19.0.1084.46 Safari/536.5'),
        ('Mozilla/5.0 (Windows; Windows NT 6.1) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.46'
         'Safari/536.5')
    )
    try:
        r = requests.get(
            url,
            timeout=12,
            headers={'User-Agent': random.choice(USER_AGENTS)}, **kwargs
        )
    except ConnectionError:
        exit_after_echo('Network connection failed.')
    except Timeout:
        exit_after_echo('timeout.')
    return r

Example #6

def get_search_results(self, video):
        results = OrderedDict()
        for i, downloader in enumerate(self.downloader):
            try:
                result = downloader.get_subtitles(video, sub_num=self.sub_num)
                results.update(result)
            except ValueError as e:
                print("error: " + str(e))
            except (exceptions.Timeout, exceptions.ConnectionError):
                print("connect timeout, search next site.")
                if i == (len(self.downloader) - 1):
                    print("PLEASE CHECK YOUR NETWORK STATUS")
                    sys.exit(0)
                else:
                    continue
            # TODO: search all sites or exit after collecting enough results
            if len(results) >= self.sub_num:
                break
        return results

Example #7

def post(self, *args, **kwargs):
        """Straightforward wrapper around `requests.Session.post
        <http://docs.python-requests.org/en/master/api/#requests.Session.post>`__.

        :return: `requests.Response
            <http://docs.python-requests.org/en/master/api/#requests.Response>`__
            object with a *soup*-attribute added by :func:`_add_soup`.
        """

        try:
            response = self.session.post(*args, **kwargs)
            self._update_state(response)
            return response
        except Timeout:
            self.timeout_exception = True
            print(f'Timeout exception.')
            resp = Response()
            resp.status_code = 408
            self._update_state(resp)
            return resp

Example #8

def test_gather_logs_failed(self, mock_post, mock_get, mock_logger):
        """SalesforceApp - Gather event logs but log files returns empty"""
        self.set_config_values(
            'CLIENT_ID', 'CLIENT_SECRET', 'USERNAME', 'PASSWORD', 'SECURITY_TOKEN'
        )
        self._app._instance_url = 'MY_URL'

        mock_post.return_value = Mock(
            status_code=200,
            json=Mock(return_value={'access_token': 'AUTH_TOKEN', 'instance_url': 'MY_URL'})
        )

        mock_get.return_value = Mock(
            status_code=200,
            json=Mock(side_effect=[list_salesforce_api_versions(), Timeout])
        )

        assert_equal(self._app._gather_logs(), None)
        mock_logger.assert_called_once()

        mock_get.return_value = Mock(
            status_code=204,
            json=Mock(return_value={'errorCode': 'ERROR_CODE', 'message': 'error message'})
        )
        assert_equal(self._app._gather_logs(), None)

Example #9

def test_do_api_call_succeeds_after_retrying(self):
        for exception in [requests_exceptions.ConnectionError,
                          requests_exceptions.SSLError,
                          requests_exceptions.Timeout,
                          requests_exceptions.ConnectTimeout,
                          requests_exceptions.HTTPError]:
            with mock.patch('airflow.providers.databricks.hooks.databricks.requests') as mock_requests:
                with mock.patch.object(self.hook.log, 'error') as mock_errors:
                    setup_mock_requests(
                        mock_requests,
                        exception,
                        error_count=2,
                        response_content={'run_id': '1'}
                    )

                    response = self.hook._do_api_call(SUBMIT_RUN_ENDPOINT, {})

                    self.assertEqual(mock_errors.call_count, 2)
                    self.assertEqual(response, {'run_id': '1'})

Example #10

def exception_handle(method):
    """Handle exception raised by requests library."""

    def wrapper(*args, **kwargs):
        try:
            result = method(*args, **kwargs)
            return result
        except ProxyError:
            LOG.exception('ProxyError when try to get %s.', args)
            raise ProxyError('A proxy error occurred.')
        except ConnectionException:
            LOG.exception('ConnectionError when try to get %s.', args)
            raise ConnectionException('DNS failure, refused connection, etc.')
        except Timeout:
            LOG.exception('Timeout when try to get %s', args)
            raise Timeout('The request timed out.')
        except RequestException:
            LOG.exception('RequestException when try to get %s.', args)
            raise RequestException('Please check out your network.')

    return wrapper

Example #11

def get_my_public_url():
    if 'PUBLIC_URL' in os.environ:
        return os.environ['PUBLIC_URL']
    try:
        if os.environ.get('PORT', '80') != '80':
            port = ':' + os.environ.get('PORT', '80')
        else:
            port = ''
        ip = get('http://ip.42.pl/raw').text
        has_public_address = get(
            f'http://{ip}{port}/ping'
        ).text == 'pong'
    except (ConnectionError, Timeout):
        return None
    if has_public_address:
        return f'http://{ip}{port}'
    else:
        return None

Example #12

def test_broadcast_block_raise_exception(
        fx_session: scoped_session, fx_user: User,
        error: typing.Union[ConnectionError, Timeout]
):
    block = Block.create(fx_user, [])
    url = 'http://test.neko'
    now = datetime.datetime.utcnow()
    node = Node(url=url, last_connected_at=now)
    fx_session.add(node)
    fx_session.flush()
    with Mocker() as m:
        m.post('http://test.neko/blocks', exc=error)
        multicast(
            serialized=block.serialize(
                use_bencode=False,
                include_suffix=True,
                include_moves=True,
                include_hash=True
            ),
            broadcast=broadcast_block,
        )
        assert node.last_connected_at == now

Example #13

def _login(self):
        """
        Login to the xqueue server
        """

        try:
            self.session = requests.Session()
            self.session.auth = (self.auth_user, self.auth_pass)
            request = self.session.post('{0}/xqueue/login/'.format(self.url),
                                        data={'username': self.queue_user,
                                              'password': self.queue_pass})
            response = json.loads(request.text)
            if response['return_code'] != 0:
                raise Exception("Invalid return code in reply resp:{0}".format(
                    str(response)))
        except (Exception, ConnectionError, Timeout) as e:
            log.critical("Unable to connect to queue xqueue: {0}".format(e))
            raise

Example #14

def get_length(self):
        """
        Returns the length of the queue
        """

        try:
            request = self.session.get('{0}/xqueue/get_queuelen/'.format(
                self.url), params={'queue_name': self.queue_name})
            response = json.loads(request.text)
            if response['return_code'] != 0:
                raise Exception("Invalid return code in reply")
            length = int(response['content'])
        except (ValueError, Exception, ConnectionError, Timeout) as e:
            log.critical("Unable to get queue length: {0}".format(e))
            raise

        return length

Example #15

def get_submission(self):
        """
        Gets a single submission from the xqueue
        server and returns the payload as a dictionary
        """

        try:
            request = self.session.get('{0}/xqueue/get_submission/'.format(
                self.url), params={'queue_name': self.queue_name})
        except (ConnectionError, Timeout) as e:
            log.critical("Unable to get submission from queue xqueue: {0}".format(e))
            raise

        try:
            response = json.loads(request.text)
            log.debug('response from get_submission: {0}'.format(response))
            if response['return_code'] != 0:
                log.critical("response: {0}".format(request.text))
                raise Exception("Invalid return code in reply")

            return json.loads(response['content'])

        except (Exception, ValueError, KeyError) as e:
            log.critical("Unable to parse xqueue message: {0} response: {1}".format(e, request.text))
            raise

Example #16

def _DoHttpRequestAsync(self, sink_stack, deadline, stream, msg):
    if deadline:
      timeout = deadline - time.time()
      if timeout < 0:
        sink_stack.AsyncProcessResponseMessage(MethodReturnMessage(error=TimeoutError()))
    else:
      timeout = None

    try:
      response = self._MakeRequest(msg, stream, timeout)
      if self._raise_on_http_error and 400 <= response.status_code < 600:
        err_text = 'HTTP Error %d: %s.' % (response.status_code, response.reason)
        if response.text:
          err_text += 'nThe server returned:n%s' % response.text
        err = exceptions.HTTPError(err_text, response=response)
        msg = MethodReturnMessage(error=err)
      else:
        self._ProcessResponse(response, sink_stack)
        return
    except exceptions.Timeout:
      msg = MethodReturnMessage(error=TimeoutError())
    except Exception as ex:
      msg = MethodReturnMessage(error=ex)
    sink_stack.AsyncProcessResponseMessage(msg)

Example #17

def check_release(self):
        loop = asyncio.get_event_loop()

        logger.info('正在检测更新...')
        try:
            resp = await loop.run_in_executor(
                None,
                partial(requests.get, 'https://pypi.org/pypi/feeluown/json', timeout=2)
            )
        except (ConnectionError, Timeout) as e:
            logger.warning(e)
            logger.warning('检查更新失败')
        else:
            rv = resp.json()
            latest = parse_version(rv['info']['version'])
            current = parse_version(__version__)
            if latest > current:
                msg = '检测到新版本 %s，当前版本为 %s' % (latest, current)
                logger.warning(msg)
                if self._app.mode & self._app.GuiMode:
                    self._app.ui.magicbox.show_msg(msg)
            else:
                logger.info('当前已经是最新版本')
                if self._app.mode & self._app.GuiMode:
                    self._app.ui.magicbox.show_msg('当前已经是最新版本')

Example #18

def get_listings(start=1,limit=1000):
    parameters = {
        'start': start,
        'limit': str(limit),
        'convert': 'USD',
        'sort': 'market_cap'
    }
    #sort=market_cap&start=1&limit=10&cryptocurrency_type=tokens&convert=USD,BTC
    headers = {
        'Accepts': 'application/json',
        'X-CMC_PRO_API_KEY': cmc_key,
    }

    session = Session()
    session.headers.update(headers)

    endpoint_summary = 'cryptocurrency/listings/latest'
    try:
        url = base_url + endpoint_summary
        response = session.get(url, params=parameters)
        data = json.loads(response.text)["data"]
        return data
    except (ConnectionError, Timeout, TooManyRedirects) as e:
        print(e)

Example #19

def get_info(idlist):
    endpoint_description = "cryptocurrency/info"
    
    #[str(x)+',' for x in range(1,20)]
    parameters = {
        #'symbol': 'BTC,ETH,XRP,LTC'
        'id':idlist
    }
    headers = {
        'Accepts': 'application/json',
        'X-CMC_PRO_API_KEY': cmc_key,
    }

    session = Session()
    session.headers.update(headers)

    try:
        url = base_url + endpoint_description
        response = session.get(url, params=parameters)
        data = json.loads(response.text)["data"]
        return data
    except (ConnectionError, Timeout, TooManyRedirects) as e:
        print("error ", e)

Example #20

def get_coin_map(active):
    endpoint_map = "cryptocurrency/map"    
    parameters = {
        'listing_status': active
    }
    headers = {
        'Accepts': 'application/json',
        'X-CMC_PRO_API_KEY': cmc_key,
    }

    session = Session()
    session.headers.update(headers)

    try:
        url = base_url + endpoint_map
        response = session.get(url, params=parameters)
        data = json.loads(response.text)["data"]
        return data
    except (ConnectionError, Timeout, TooManyRedirects) as e:
        print(e)

Example #21

def exchange_map():    
    endpoint_map = "exchange/map"
    parameters = {
        #'listing_status': active
    }
    #limit
    #slug
    headers = {
        'Accepts': 'application/json',
        'X-CMC_PRO_API_KEY': cmc_key,
    }

    session = Session()
    session.headers.update(headers)

    try:
        url = base_url + endpoint_map
        response = session.get(url, params=parameters)
        data = json.loads(response.text)
        return data
    except (ConnectionError, Timeout, TooManyRedirects) as e:
        print(e)

Example #22

def download(self, publication):
        """
        Download a publication from the mirror to the current directory.

        :param publication: a Publication
        """
        for (n, mirror) in publication.mirrors.items():
            # print(f"About to try {n}n")
            try:
                mirror.download_publication(publication)
                break  # stop if successful
            except (CouldntFindDownloadUrl, Timeout) as e:
                print(e)
                print("Trying a different mirror.")
                continue
            except Exception:
                import traceback
                print(f"An error occurred: {sys.exc_info()[0]}")
                print(traceback.format_exc())
                print("Trying a different mirror.")
                continue
            print("Failed to download publications.")

Example #23

def _spider_run(self, url):
        """ 执行真正的请求。控制代理， 超时等设置。。"""
        p = None
        try_times = 0

        while True:
            try:
                if self.method == 'post':
                    resp = self.pspider.brower.post(url, timeout=self.timeout, params=self.postdata)
                elif self.method == 'get':
                    resp = self.pspider.brower.get(url, timeout=self.timeout, **self.kw)
                else:
                    raise SpiderException(SpiderException.DEFAULT, "不支持其它方法")
                log.info("请求URL={}".format(url))
                log.info("响应字段长度={}".format(len(resp.content)))
                return resp
            except (Timeout, ConnectionError):
                self.pspider.reset_brower()
                try_times += 1
                log.info("重试 ip={} url={} retry={}".format(p, url, try_times))
                if try_times >= self.retry:
                    log.info("超过重试次数 ip={} url={}".format(p, url))
                    break

Example #24

def _fetch_data(self, urls):
        # we only want warning+ messages from the requests module
        logging.getLogger("requests").setLevel(logging.WARNING)
        for url in urls:
            try:
                req = requests.get(url, timeout=3)
                if req.status_code == 200:
                    data = req.text.strip()
                    if data is None or not self._verify_address(data):
                        continue
                    else:
                        return data
                else:
                    raise ConnectionError
            except (Timeout, ConnectionError) as e:
                logger.warning('Could not fetch public ip from %s', url)
        return None

Example #25

def import_offering_group(self, slugs):
    offerings = CourseOffering.objects.filter(slug__in=slugs)
    for o in offerings:
        logger.debug('Importing %s' % (o.slug,))
        try:
            importer.import_offering_members(o)
        except Timeout as exc:
            # elasticsearch timeout: have celery pause while it collects it thoughts, and retry
            raise self.retry(exc=exc)

Example #26

def confirm(self):
        url=self.le_input.text()
        try:
            self.adds,self.title,self.infos=get_video(url)
            self.show_cover(self.infos['cover'])
            self.show_infos()
            self.update_cb()
            self.cb_quality.setCurrentIndex(0)
            self.show_url()
            
        except (Timeout,HTTPError,ConnectionError):
            exc=traceback.format_exc()
            QtWidgets.QMessageBox.warning(self,'Warning','Got exception:n'+exc)
            self.tb_infos.setText(exc)

Example #27

def main():
    url = sys.argv[1]
    mdk = start()
    session = mdk.session()
    session.setDeadline(1.0)
    mdk.stop()
    req_ssn = requests_session(session)
    try:
        req_ssn.get(url).content
    except Timeout:
        sys.exit(123)
        return

Example #28

def post(self, url, params,headers):#post消息
        data = json.dumps(params)
        try:
            self.r =requests.post(url,params=data,headers=headers,timeout=Interface_Time_Out)
            json_response = json.loads(self.r.text)
            spend=self.r.elapsed.total_seconds()
            return json_response,spend
        except exceptions.Timeout :
            return {'post请求出错': "请求超时" }
        except exceptions.InvalidURL:
            return {'post请求出错': "非法url"}
        except exceptions.HTTPError:
            return {'post请求出错': "http请求错误"}
        except Exception as e:
            return {'post请求出错': "错误原因:%s" % e}

Example #29

def delfile(self,url,params,headers):#删除的请求
        try:
            self.rdel_word=requests.delete(url,data=params,headers=headers,timeout=Interface_Time_Out)
            json_response=json.loads(self.rdel_word.text)
            spend=self.rdel_word.elapsed.total_seconds()
            return json_response,spend
        except exceptions.Timeout :
            return {'delete请求出错': "请求超时" }
        except exceptions.InvalidURL:
            return {'delete请求出错': "非法url"}
        except exceptions.HTTPError:
            return {'delete请求出错': "http请求错误"}
        except Exception as e:
            return {'delete请求出错': "错误原因:%s" % e}

Example #30

def putfile(self,url,params,headers):#put请求
        try:
            self.rdata=json.dumps(params)
            me=requests.put(url,self.rdata,headers=headers,timeout=Interface_Time_Out)
            json_response=json.loads(me.text)
            spend=me.elapsed.total_seconds()
            return json_response,spend
        except exceptions.Timeout :
            return {'put请求出错': "请求超时" }
        except exceptions.InvalidURL:
            return {'put请求出错': "非法url"}
        except exceptions.HTTPError:
            return {'put请求出错': "http请求错误"}
        except Exception as e:
            return {'put请求出错': "错误原因:%s" % e}
Источник

In this tutorial, you’ll learn how to use timeouts in the Python requests library, when working with any type of HTTP request being made. By default, the requests library will not time out any request you make, which can result in your programming running indefinitely if a server doesn’t respond.

By the end of this tutorial, you’ll have learned:

  • How to set timeouts in requests
  • How to set unique timeouts for connecting and reading in Python requests
  • How to catch and handle timeout errors in Python requests

How Does Python requests Handle Timeouts?

By default, the Python requests library does not set a timeout for any request it sends. This is true for GET, POST, and PUT requests. While this can prevent unexpected errors, it can result in your request running indefinitely.

Because of this, it’s important to set a timeout to prevent unexpected behavior. Remember, the Python requests library will not timeout by default, unless explicitly instructed.

How to Set a Timeout for Python requests

In order to set a timeout in an HTTP request made via the requests library, you can use the timeout parameter. The parameter accepts either an integer or a floating point value, which describes the time in seconds.

It’s important to note that this behavior is different from many other HTTP request libraries, such as those in JavaScript. In other libraries or languages, this behavior tends to be expressed in milliseconds.

Let’s take a look at an example of how we can send a GET request with a timeout:

# Setting a Timeout on a GET Request with an Integer
import requests
resp = requests.get('https://datagy.io', timeout=3)

In the example above, we set a timeout of 3 seconds. We used an integer to represent the time of our timeout. If we wanted to be more precise, we could also pass in a floating point value:

# Setting a Timeout on a GET Request with a Floating Point Value
import requests
resp = requests.get('https://datagy.io', timeout=3.5)

By passing in a single value, we set a timeout for the request. If we wanted to set different timeouts for connecting and reading a request, we can pass in a tuple of values.

How to Set Timeouts for Connecting and Reading in Python requests

In some cases, you’ll want to set different timeouts for making a connection and for reading results. This can easily be done using the timeout parameter in the requests library. Similar to the example above, this can be applied to any type of request being made.

Let’s see how we can pass in different timeout limits for connecting and reading requests in the Python requests library:

# Setting Different Timeouts for Connecting and Reading Requests
import requests
resp = requests.get('https://datagy.io', timeout=(1, 2))

In the example above, we set the request to timeout after 1 second for connecting and 2 seconds for reading the request.

In the following section, you’ll learn how to catch and handle errors that arise due to requests timing out.

How to Catch and Handle Timeout Errors in Python requests

When applying a timeout, it’s important to note that this is not a time limit on the entire response. Instead, it raises an exception if no bytes have been received on the underlying socket.

If the request does not receive any bytes within the specified timeout limit, a Timeout error is raised. Let’s see what this looks like:

# Raising a Timeout Error in a Python requests GET Request
import requests
resp = requests.get('https://datagy.io', timeout=0.0001)

# Raises:
# ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x7fbc988f59a0>, 'Connection to datagy.io timed out. (connect timeout=0.0001)')

In order to prevent your program from crashing, you need to handle the exception using a try-except block. Let’s see how this can be done:

# Handling a Timeout Error
import requests
from requests.exceptions import ConnectTimeout

try:
    requests.get('https://datagy.io', timeout=0.0001)
except ConnectTimeout:
    print('Request has timed out')

# Returns:
# Request has timed out

We can see in the code above that the error was handled safely. In order to do this, we:

  1. Imported the error from the exceptions module of the requests library
  2. We created a try-except block to handle the ConnectTimeout error.

Frequently Asked Questions

What is the default timeout for Python requests?

None. There is no default timeout for Python requests, unless explicitly set using the timeout parameter.

How do you set a timeout for requests made in Python?

You set a timeout (in seconds) using the timeout= parameter when making HTTP requests in the Python requests library.

What is the best time to set for a timeout for requests made in Python?

While there is no best set value for timeouts for HTTP requests made in Python, a good practice is to set them under 500ms. This allows your application to provide a better user experience and to process more requests.

Conclusion

In this tutorial, you learned how to handle timeouts in the Python requests library. You first learned how the Python requests library handles timeouts. Then, you learned how to set timeouts when making HTTP requests, both using integers and floating point values. Then, you learned how to specify specific timeouts for connecting and reading requests. Finally, you learned how to handle timeout exceptions in the Python requests library.

Additional Resources

To learn more about related topics, check out the tutorials below:

  • Python Requests Response Object Explained
  • Python Requests Headers Explained
  • Python Requests Sessions Explained
  • Official Documentation: Python requests Timeout
Источник

The most usual way of making HTTP/HTTPS requests in Python applications is using the requests library. By default it doesn’t define any timeouts for the operations. It means that in specific circumstances a simple requests.get() call might not return at all. Depending on the nature of the application this is not desirable. This post demonstrates different ways to handle those situations.

But first, let’s see how it actually looks like when we don’t have timeout configured. I have two examples here (the statements have been pasted from a text editor so there is no user input latency in the time calculation), the first one:

>>> import time
>>> import requests
>>> start_time = time.monotonic()
>>> try:
...     r = requests.get("http://192.168.7.11/")
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectionError'>
HTTPConnectionPool(host='192.168.7.11', port=80): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f2bd655eb20>: Failed to establish a new connection: [Errno 113] No route to host'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
3.08 seconds
>>>

Captured packets for the first attempt:

Another:

>>> start_time = time.monotonic()
>>> try:
...     r = requests.get("http://192.168.8.1/")
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectionError'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f2bd65a48e0>: Failed to establish a new connection: [Errno 110] Connection timed out'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
130.41 seconds
>>>

Captured packets for the second attempt:

As we can see, without any timeout defined in the requests.get() call there definitely was timeout involved.

Note: The test host here was a Debian 11 Bullseye server, with a Linux kernel from 5.10 series. With another operating system/kernel the results can/will be different.

What actually happened is that requests used services from urllib3, which used services from http.client, which in turn used services from socket. And it was the socket.connect() call that down there returned an error when error/nothing happened with the connection attempts.

The first example was about connection attempt to a host in the local subnet but the destination host did not exist, thus no ARP response. The socket call timed out already in three seconds (even though the host networking stack continued attempting the ARP requests for another two seconds).

The second example was a remote destination that was not responding because the destination IP was incorrect or firewall blocked the requests, or some other error happened that prevented any response. In this case the TCP stack did its usual attempts with exponentially increased delay between the attempts (1, 2, 4, 8, 16 and ~32 seconds). (Btw: No ARP request/response was shown because the host already had the local router MAC address in the cache.) And finally the stack gave up at about 130 seconds of total waiting. I also run this same scenario on a Debian 10 Buster server (with kernel 4.19) and there the timeout happened at about 30 seconds.

Let’s try adjusting the timeout now with the basic timeout argument in requests.get():

>>> start_time = time.monotonic()
>>> try:
...     r = requests.get("http://192.168.7.11/", timeout=2)
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.7.11', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd656b070>, 'Connection to 192.168.7.11 timed out. (connect timeout=2)'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
2.01 seconds
>>>

And with the remote destination:

>>> start_time = time.monotonic()
>>> try:
...     r = requests.get("http://192.168.8.1/", timeout=2)
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd6556d30>, 'Connection to 192.168.8.1 timed out. (connect timeout=2)'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
2.0 seconds
>>>

In both the local subnet and remote cases the raised exception was changed from requests.ConnectionError to requests.ConnectTimeout, and the timeout happened in two seconds, just like expected.

Actually the timeout argument is a bit more complex: you can specify a tuple of two values. The first value is the connect timeout and the second is the read timeout. Read the requests advanced documentation for more about that detail.

Extra option: Whenever using the timeout argument, instead of using the int/float values you can use a urllib3.Timeout instance, like this:

>>> from urllib3 import Timeout
>>> start_time = time.monotonic()
>>> try:
...     r = requests.get("http://192.168.8.1/", timeout=Timeout(connect=2, read=60))
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd6556910>, 'Connection to 192.168.8.1 timed out. (connect timeout=2)'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
2.01 seconds
>>>

I don’t know if there are other reasons for using the Timeout object instead of tuple of values than to make the configuration more visual: timeout=Timeout(connect=x, read=y) instead of timeout=(x, y).

Ok, now we know how to specify the timeout for separate requests calls. Now it’s just all about remembering to add it to every call separately… maybe not. Instead, we can use sessions and transport adapters to set a timeout for all calls:

>>> from requests.adapters import HTTPAdapter
>>> class TimeoutHTTPAdapter(HTTPAdapter):
...     def __init__(self, *args, **kwargs):
...         if "timeout" in kwargs:
...             self.timeout = kwargs["timeout"]
...             del kwargs["timeout"]
...         else:
...             self.timeout = 5   # or whatever default you want
...         super().__init__(*args, **kwargs)
...     def send(self, request, **kwargs):
...         kwargs["timeout"] = self.timeout
...         return super().send(request, **kwargs)
...
>>> session = requests.Session()
>>> adapter = TimeoutHTTPAdapter(timeout=(1,3))
>>> session.mount("http://", adapter)
>>> session.mount("https://", adapter)
>>> start_time = time.monotonic()
>>> try:
...     r = session.get("http://192.168.8.1/")
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd655ebe0>, 'Connection to 192.168.8.1 timed out. (connect timeout=1)'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
1.01 seconds
>>>

The instance of the custom TimeoutHTTPAdapter class was mounted to all http:// and https:// URLs with the session object, and then the GET request was specifically executed via the session object (session.get()), not via the requests.get() global function.

What the TimeoutHTTPAdapter class actually does is it overrides any call-specific timeout arguments with the instance-defined timeout attribute when sending requests. Here the timeout=4 argument is effectively ignored because of that:

>>> try:
...     r = session.get("http://192.168.8.1/", timeout=4)
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd650bf40>, 'Connection to 192.168.8.1 timed out. (connect timeout=1)'))
>>>

If you will, you can edit the TimeoutHTTPAdapter.send() method to support call-level overrides:

>>> class TimeoutHTTPAdapter(HTTPAdapter):
...     def __init__(self, *args, **kwargs):
...         if "timeout" in kwargs:
...             self.timeout = kwargs["timeout"]
...             del kwargs["timeout"]
...         else:
...             self.timeout = 5   # or whatever default you want
...         super().__init__(*args, **kwargs)
...     def send(self, request, **kwargs):
...         if kwargs["timeout"] is None:
...             kwargs["timeout"] = self.timeout
...         return super().send(request, **kwargs)
...
>>> session = requests.Session()
>>> adapter = TimeoutHTTPAdapter(timeout=(1,3))
>>> session.mount("http://", adapter)
>>> session.mount("https://", adapter)
>>> start_time = time.monotonic()
>>> try:
...     r = session.get("http://192.168.8.1/", timeout=4)
... except Exception as e:
...     print(type(e))
...     print(e)
...
<class 'requests.exceptions.ConnectTimeout'>
HTTPConnectionPool(host='192.168.8.1', port=80): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f2bd64f5bb0>, 'Connection to 192.168.8.1 timed out. (connect timeout=4)'))
>>> stop_time = time.monotonic()
>>> print(round(stop_time-start_time, 2), "seconds")
4.01 seconds
>>>

In this case the session-level timeout value is used in send() only when the call-level timeout is None (which it is by default).

Conclusion

Use a custom requests.adapters.HTTPAdapter class (like TimeoutHTTPAdapter in this post) to specify your application-level timeout values.

In the next post we will see request retries in action.

Источник
  1. Implement Timeout for a URL in Python
  2. Implement Timeout on Multiple URLs Using the Requests Module in Python
  3. Set None Inside Timeout Parameter to Implement Timeout in Python Requests
  4. Implement Timeout for an Alternative to Python Requests
  5. Conclusion

Implement Timeout in Python Requests

This article addresses how to implement timeouts in Python requests.

Timeouts must put a timer in request.get() while fetching data from a connection. If a connection takes more time than the required threshold inside a timeout parameter, request.get() gets timed out.

The examples below explain different methods of implementing timeout for Python requests using the syntax requests.get().

Implement Timeout for a URL in Python

This program demonstrates the method to fetch a response from an URL using GET and implementing timeout inside it.

  • Import the Python library package requests.
  • Define a try block.
  • Inside the try block, declare a variable req to store requests from the required URL and set the timeout parameter.
  • After the timeout parameter is set, print req to view the response.
  • Inside the except block, set an exception if the program does not receive any response and print the message.

The try-except block is an exception handling block that lets the program execute a function and sets an alternative if an exception arises.

Here, the program connects to the URL https://www.google.com and executes the GET command. The GET command fetches a response from the connection, which is usually <Response [200]> for successful connections.

The syntax timeout=(1) tells the program to timeout connection after 1 second if no response is received.

Example:

import requests

try:
    req = requests.request('GET', 'https://www.google.com',timeout=(1))
    print(req)
except requests.ReadTimeout:
    print("READ TIME OUT")

Output:

"C:UsersWin 10main.py"
<Response [200]>

Process finished with exit code 0

Implement Timeout on Multiple URLs Using the Requests Module in Python

This example demonstrates the method to fetch the response from multiple URLs simultaneously. In this program, along with fetching the response, the timeout parameter will be implemented using a tuple; for example, (x,y) is a tuple.

A timeout can be set as a tuple in reading and connecting, specified separately inside the program.

A timeout=(1,3) indicates a connect timer of 1 second and a read timer of 3 seconds.

It must be noted that the timeout given will be applied to all the URLs. If different URLs need different timeouts, the program should contain the request.get() function for the number of times various timeouts are there.

  • Import the library packages requests and urllib3.

    urllib3 handles exceptions that arise from internet firewalls when the program tries to connect to domains the firewall does not recognize.

    The syntax below disables the warnings which arise when the program connects to an insecure website.

    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

  • Inside the try block, initialize the variable websites to store multiple domains.

  • Run a for loop w for the number of objects present inside the variable website.

  • Initialize variable r to store request.get() response and specify timeout (3,3). Verify checks for the website TLS certificate.

    It is given a false value to avoid exception throws.

  • Print variable r.

  • Inside the except block, store the timeout exception inside e and print it.

    Example:

    import requests as requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
    websites=['https://github.com','https://www.google.com', 'https://1337xto.to']
    for w in websites:
        r = requests.get(w, verify=False, timeout=(3,3))
        print(r)

except requests.exceptions.Timeout as e:
    print(e)

Warning

The example codes in this artilce are susceptible to the security vulnerability — Improper Certificate Validation. The software does not validate or incorrectly validates a certificate. This might allow an attacker to spoof a trusted server or act as a man-in-the-middle by interfering in the communication path between the host and client. The attacker might be able to intercept sensitive data or send data that appear to originate from a trusted server. (CWE-295)

The program tries to connect to three URLs. After successfully connecting to the first two, the program prints the response.

In the third URL, the program throws a timeout exception as the URL takes more than 6 seconds to return a response.

Output:

"C:UsersWin 10main.py"
<Response [200]>
<Response [200]>
HTTPSConnectionPool(host='1337xto.to', port=443): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x0000022B3A202940>, 'Connection to 1337xto.to timed out. (connect timeout=3)'))

Process finished with exit code 0

Set None Inside Timeout Parameter to Implement Timeout in Python Requests

In a particular scenario where the connection is made to a very slow website, and the response time takes more than it usually takes, timeouts are set to None.

Example:

import requests

try:
    req = requests.request('GET', 'https://1337xto.to', timeout=None)
    print(req)

except requests.ReadTimeout:
    print("READ TIME OUT")

Implement Timeout for an Alternative to Python Requests

There are many alternatives for Python requests’ request.get() timeout, though most are redundant or unsupported. A third-party library package eventlet is used to execute timeout in this example.

  • Import Python library packages — requests and eventlet.
  • monkey_patch patches the standard eventlet library with its green equivalents.
  • Inside try block, set eventlet.Timeout for desired seconds.
  • Initialize variable req to store URL response using request.get() syntax.
  • Inside the except block, print timeout message.

Example:

import requests
import eventlet
eventlet.monkey_patch()

try:
    with eventlet.Timeout(10):
        req = requests.get("http://ipv4.download.thinkbroadband.com/1GB.zip", verify=False)
        print(req)

except:
    print('timeout')

Output:

"C:UsersWin 10curl.py"
timeout

Process finished with exit code 0

Conclusion

Implementing timeout in request.get() is explained with three example codes of Python requests, and another example is given to demonstrate timeout in the eventlet.

After going through this article, the reader will be able to implement a timeout in Python requests.

Источник

Понравилась статья? Поделить с друзьями:
  • Python try except get error message
  • Python try except error code
  • Python try except any error
  • Python tkinter окно ошибки
  • Python throw error