Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/utils.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019-2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import copy | import copy | ||||
| import functools | import functools | ||||
| import logging | import logging | ||||
| import os | import os | ||||
| from typing import Callable, Dict, Optional, Tuple, TypeVar | from typing import Callable, Dict, Optional, Tuple, TypeVar | ||||
| import requests | import requests | ||||
| from swh.loader.exception import NotFound | |||||
| from swh.loader.package import DEFAULT_PARAMS | from swh.loader.package import DEFAULT_PARAMS | ||||
| from swh.model.hashutil import HASH_BLOCK_SIZE, MultiHash | from swh.model.hashutil import HASH_BLOCK_SIZE, MultiHash | ||||
| from swh.model.model import Person | from swh.model.model import Person | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| DOWNLOAD_HASHES = set(["sha1", "sha256", "length"]) | DOWNLOAD_HASHES = set(["sha1", "sha256", "length"]) | ||||
| EMPTY_AUTHOR = Person(fullname=b"", name=None, email=None,) | EMPTY_AUTHOR = Person(fullname=b"", name=None, email=None,) | ||||
| def api_info(url: str, **extra_params) -> bytes: | def api_info(url: str, **extra_params) -> bytes: | ||||
| """Basic api client to retrieve information on project. This deals with | """Basic api client to retrieve information on project. This deals with | ||||
| fetching json metadata about pypi projects. | fetching json metadata about pypi projects. | ||||
| Args: | Args: | ||||
| url (str): The api url (e.g PyPI, npm, etc...) | url (str): The api url (e.g PyPI, npm, etc...) | ||||
| Raises: | Raises: | ||||
| ValueError in case of query failures (for some reasons: 404, ...) | NotFound in case of query failures (for some reasons: 404, ...) | ||||
| Returns: | Returns: | ||||
| The associated response's information | The associated response's information | ||||
| """ | """ | ||||
| response = requests.get(url, **{**DEFAULT_PARAMS, **extra_params}) | response = requests.get(url, **{**DEFAULT_PARAMS, **extra_params}) | ||||
| if response.status_code != 200: | if response.status_code != 200: | ||||
| raise ValueError("Fail to query '%s'. Reason: %s" % (url, response.status_code)) | raise NotFound(f"Fail to query '{url}'. Reason: {response.status_code}") | ||||
| return response.content | return response.content | ||||
| def download( | def download( | ||||
| url: str, | url: str, | ||||
| dest: str, | dest: str, | ||||
| hashes: Dict = {}, | hashes: Dict = {}, | ||||
| filename: Optional[str] = None, | filename: Optional[str] = None, | ||||
| ▲ Show 20 Lines • Show All 92 Lines • Show Last 20 Lines | |||||