Fetch `url` and return the temporary location where the fetched content was saved. Use `file_name` if provided or create a new `file_name` base on the last url segment. If `verify` is True, SSL certification is performed. Otherwise, no verification is done but a warning will be prin
(url, file_name=None, verify=True, timeout=10)
| 23 | |
| 24 | |
| 25 | def download_url(url, file_name=None, verify=True, timeout=10): |
| 26 | """ |
| 27 | Fetch `url` and return the temporary location where the fetched content was |
| 28 | saved. Use `file_name` if provided or create a new `file_name` base on the last |
| 29 | url segment. If `verify` is True, SSL certification is performed. Otherwise, no |
| 30 | verification is done but a warning will be printed. |
| 31 | `timeout` is the timeout in seconds. |
| 32 | """ |
| 33 | requests_args = dict(timeout=timeout, verify=verify) |
| 34 | file_name = file_name or fileutils.file_name(url) |
| 35 | |
| 36 | try: |
| 37 | response = requests.get(url, **requests_args) |
| 38 | except (ConnectionError, InvalidSchema) as e: |
| 39 | logger.error("download_url: Download failed for %(url)r" % locals()) |
| 40 | raise |
| 41 | |
| 42 | status = response.status_code |
| 43 | if status != 200: |
| 44 | msg = "download_url: Download failed for %(url)r with %(status)r" % locals() |
| 45 | logger.error(msg) |
| 46 | raise Exception(msg) |
| 47 | |
| 48 | tmp_dir = fileutils.get_temp_dir(prefix="fetch-") |
| 49 | output_file = os.path.join(tmp_dir, file_name) |
| 50 | with open(output_file, "wb") as out: |
| 51 | out.write(response.content) |
| 52 | |
| 53 | return output_file |
| 54 | |
| 55 | |
| 56 | def ping_url(url): |
nothing calls this directly
no test coverage detected