Fetch a URL using requests library Args: url (str): URL to fetch content from max_retries (int, optional): Maximum number of retry attempts backoff_factor (float, optional): A backoff factor to apply between attempts timeout (int, optional): How many seconds to w
(url, max_retries=3, backoff_factor=0.5, timeout=60,
headers=None)
| 134 | |
| 135 | |
| 136 | def fetch_url(url, max_retries=3, backoff_factor=0.5, timeout=60, |
| 137 | headers=None): |
| 138 | """Fetch a URL using requests library |
| 139 | |
| 140 | Args: |
| 141 | url (str): URL to fetch content from |
| 142 | max_retries (int, optional): Maximum number of retry attempts |
| 143 | backoff_factor (float, optional): A backoff factor to apply between attempts |
| 144 | timeout (int, optional): How many seconds to wait for the server before giving up |
| 145 | headers (dict, optional): Extra HTTP headers appended to default headers |
| 146 | |
| 147 | Returns: |
| 148 | requests.Response: Response object from requests library |
| 149 | |
| 150 | Raises: |
| 151 | requests.RequestException: If there is an error fetching the URL |
| 152 | """ |
| 153 | assert isinstance(url, str) |
| 154 | assert url.startswith('http'), f"URL must start with http, got {url}" |
| 155 | assert isinstance(max_retries, int) |
| 156 | assert max_retries >= 0 |
| 157 | assert isinstance(backoff_factor, float) |
| 158 | assert backoff_factor > 0 |
| 159 | assert isinstance(timeout, int) |
| 160 | assert timeout > 0 |
| 161 | if hasattr(sys, 'frozen'): |
| 162 | # when frozen by py2exe, certificates are in alternate location |
| 163 | ca_bundle = os.path.join(bleachbit_exe_path, 'cacert.pem') |
| 164 | if os.path.exists(ca_bundle): |
| 165 | requests.utils.DEFAULT_CA_BUNDLE_PATH = ca_bundle |
| 166 | requests.adapters.DEFAULT_CA_BUNDLE_PATH = ca_bundle |
| 167 | else: |
| 168 | logger.error( |
| 169 | 'Application is frozen but certificate file not found: %s', ca_bundle) |
| 170 | assert headers is None or isinstance(headers, dict) |
| 171 | request_headers = {'User-Agent': get_user_agent()} |
| 172 | if headers: |
| 173 | request_headers.update(headers) |
| 174 | unset_sslkeylogfile(True) |
| 175 | # 408: request timeout |
| 176 | # 429: too many requests |
| 177 | # 500: internal server error |
| 178 | # 502: bad gateway |
| 179 | # 503: service unavailable |
| 180 | # 504: gateway_timeout |
| 181 | status_forcelist = (408, 429, 500, 502, 503, 504) |
| 182 | with requests.Session() as session: |
| 183 | if HAVE_URLLIB3: |
| 184 | retries = Retry(total=max_retries, backoff_factor=backoff_factor, |
| 185 | status_forcelist=status_forcelist, redirect=5) |
| 186 | session.mount( |
| 187 | 'http://', requests.adapters.HTTPAdapter(max_retries=retries)) |
| 188 | session.mount( |
| 189 | 'https://', requests.adapters.HTTPAdapter(max_retries=retries)) |
| 190 | response = session.get(url, headers=request_headers, |
| 191 | timeout=timeout, verify=True) |
| 192 | return response |
| 193 |