Coverage for rust2rpm/cratesio.py: 89%

1"""Module containing functionality for interacting with the crate registry on <https://crates.io>.""" 

2 

3import contextlib 

4import os 

5from pathlib import Path 

6from urllib.parse import urljoin 

7 

8import requests 

9import tqdm 

10from cargo2rpm.semver import Version 

11 

12from rust2rpm import log 

13 

14CRATES_IO_API_URL = "https://crates.io/api/v1/" 

15"""The root of all URLs of the crates.io API.""" 

16 

17if XDG_CACHE_HOME_STR := os.getenv("XDG_CACHE_HOME"): 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 XDG_CACHE_HOME = Path(XDG_CACHE_HOME_STR) 

19else: 

20 XDG_CACHE_HOME = Path("~/.cache").expanduser() 

21 

22CACHE_DIR = XDG_CACHE_HOME / "rust2rpm" 

23"""Location of the rust2rpm download cache. 

24 

25This is where crate files that are downloaded from <https://crates.io> are 

26stored. 

27""" 

28 

29 

30class NoVersionsError(Exception): 

31 """Raised when querying crates.io yields no valid versions.""" 

32 

33 

34def query_available_versions( 

35 crate: str, 

36 *, 

37 stable: bool = True, 

38) -> list[Version]: # pragma nocover: requires internet access 

39 """Query the crates.io API for all available versions of a crate. 

40 

41 Arguments: 

42 crate: Name of the crate in the registry. 

43 stable: Filter out pre-releases if set to `True`. 

44 

45 Returns: 

46 List of available versions. 

47 

48 """ 

49 url = urljoin(CRATES_IO_API_URL, f"crates/{crate}/versions") 

50 req = requests.get(url, headers={"User-Agent": "rust2rpm"}, timeout=10) 

51 req.raise_for_status() 

52 versions = req.json()["versions"] 

53 

54 parsed_versions = (Version.parse(x["num"]) for x in filter(lambda x: not x["yanked"], versions)) 

55 

56 if stable: 

57 return list(filter(lambda x: x.pre is None, parsed_versions)) 

58 return list(parsed_versions) 

59 

60 

61def query_newest_version(crate: str) -> Version: # pragma nocover: requires internet access 

62 """Query the crates.io API for the greatest available version of a crate. 

63 

64 Arguments: 

65 crate: Name of the crate in the registry. 

66 

67 Returns: 

68 The greatest available version. 

69 

70 Raises: 

71 `NoVersionsError`: No version is available or is considered suitable. 

72 

73 """ 

74 url = urljoin(CRATES_IO_API_URL, f"crates/{crate}/versions") 

75 req = requests.get(url, headers={"User-Agent": "rust2rpm"}, timeout=10) 

76 req.raise_for_status() 

77 versions = req.json()["versions"] 

78 

79 def is_stable(s: dict) -> bool: 

80 return Version.parse(s["num"]).pre is None 

81 

82 def is_not_yanked(s: dict) -> bool: 

83 return not s["yanked"] 

84 

85 # return the most recent, non-yanked stable version 

86 

87 def is_not_yanked_and_stable(s: dict) -> bool: 

88 return is_stable(s) and is_not_yanked(s) 

89 

90 not_yanked_and_stable = [*filter(is_not_yanked_and_stable, versions)] 

91 if len(not_yanked_and_stable) > 0: 

92 return Version.parse(not_yanked_and_stable[0]["num"]) 

93 

94 # there are no non-yanked stable versions: 

95 # fall back to the latest pre-release 

96 

97 not_yanked = [*filter(is_not_yanked, versions)] 

98 if len(not_yanked) > 0: 

99 version = not_yanked[0]["num"] 

100 log.warn(f"No stable versions available. Falling back to the latest pre-release version {version!r}.") 

101 return Version.parse(version) 

102 

103 # there are no non-yanked versions: fatal 

104 msg = f"No versions are available for crate {crate!r}." 

105 raise NoVersionsError(msg) 

106 

107 

108@contextlib.contextmanager 

109def remove_on_error(path: Path): # pragma nocover: only used in download_crate 

110 """Context manager that removes the path on any error.""" 

111 try: 

112 yield 

113 except: # this is supposed to include ^C 

114 path.unlink() 

115 raise 

116 

117 

118def download_crate(name: str, version: Version, *, offline: bool) -> Path: # pragma nocover: requires internet access 

119 """Download the specified crate version from crates.io. 

120 

121 Arguments: 

122 name: Name of the crate in the registry. 

123 version: Version of the crate that is downloaded. 

124 offline: Skip download and only construct file path if set to `True`. 

125 

126 Returns: 

127 Path to the file downloaded into the download cache directory. 

128 

129 """ 

130 CACHE_DIR.mkdir(parents=True, exist_ok=True) 

131 

132 crate_file_name = f"{name}-{version}.crate" 

133 crate_file_path = CACHE_DIR / crate_file_name 

134 

135 if not crate_file_path.is_file() and not offline: 

136 url = urljoin(CRATES_IO_API_URL, f"crates/{name}/{version}/download#") 

137 req = requests.get(url, stream=True, headers={"User-Agent": "rust2rpm"}, timeout=10) 

138 req.raise_for_status() 

139 total = int(req.headers["Content-Length"]) 

140 

141 with remove_on_error(crate_file_path), crate_file_path.open("wb") as f: 

142 for chunk in tqdm.tqdm( 

143 req.iter_content(), 

144 f"Downloading {crate_file_name}", 

145 total=total, 

146 unit="B", 

147 unit_scale=True, 

148 ): 

149 f.write(chunk) 

150 

151 return crate_file_path