Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/utils.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 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 logging | import logging | ||||
| import os | import os | ||||
| import requests | import requests | ||||
| from typing import Dict, Optional, Tuple | from typing import Callable, Dict, Optional, Tuple, TypeVar | ||||
| from swh.model.hashutil import MultiHash, HASH_BLOCK_SIZE | from swh.model.hashutil import MultiHash, HASH_BLOCK_SIZE | ||||
| from swh.model.model import Person | from swh.model.model import Person | ||||
| from swh.loader.package import DEFAULT_PARAMS | from swh.loader.package import DEFAULT_PARAMS | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| ▲ Show 20 Lines • Show All 96 Lines • ▼ Show 20 Lines | ) -> Tuple[str, Dict]: | ||||
| return filepath, extrinsic_metadata | return filepath, extrinsic_metadata | ||||
| def release_name(version: str, filename: Optional[str] = None) -> str: | def release_name(version: str, filename: Optional[str] = None) -> str: | ||||
| if filename: | if filename: | ||||
| return "releases/%s/%s" % (version, filename) | return "releases/%s/%s" % (version, filename) | ||||
| return "releases/%s" % version | return "releases/%s" % version | ||||
| TReturn = TypeVar("TReturn") | |||||
| TSelf = TypeVar("TSelf") | |||||
| _UNDEFINED = object() | |||||
| def cached_method(f: Callable[[TSelf], TReturn]) -> Callable[[TSelf], TReturn]: | |||||
| cache_name = f"_cached_{f.__name__}" | |||||
| @functools.wraps(f) | |||||
| def newf(self): | |||||
| value = getattr(self, cache_name, _UNDEFINED) | |||||
| if value is _UNDEFINED: | |||||
| value = f(self) | |||||
| setattr(self, cache_name, value) | |||||
| return value | |||||
| return newf | |||||