Coverage for rust2rpm/cratesio.py: 24%

1import os 

2import re 

3from typing import Optional 

4from urllib.parse import urljoin 

5 

6import requests 

7import tqdm 

8 

9from cargo2rpm.semver import Version 

10 

11from rust2rpm import log 

12from rust2rpm.utils import remove_on_error 

13 

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

15XDG_CACHE_HOME = os.getenv("XDG_CACHE_HOME", os.path.expanduser("~/.cache")) 

16CACHE_DIR = os.path.join(XDG_CACHE_HOME, "rust2rpm") 

17 

18 

19class NoVersionsError(Exception): 

20 pass 

21 

22 

23def query_available_versions(crate: str, stable: bool = True) -> list[Version]: 

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

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

26 req.raise_for_status() 

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

28 

29 parsed_versions = map(lambda x: Version.parse(x["num"]), filter(lambda x: not x["yanked"], versions)) 

30 

31 if stable: 

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

33 else: 

34 return list(parsed_versions) 

35 

36 

37def query_newest_version(crate: str) -> str: 

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

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

40 req.raise_for_status() 

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

42 

43 def is_stable(s): 

44 return not re.search("alpha|beta|rc|pre", s["num"]) 

45 

46 def is_not_yanked(s): 

47 return not s["yanked"] 

48 

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

50 

51 def is_not_yanked_and_stable(s): 

52 return is_stable(s) and is_not_yanked(s) 

53 

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

55 if len(not_yanked_and_stable) > 0: 

56 version = not_yanked_and_stable[0]["num"] 

57 return version 

58 

59 # there are no non-yanked stable versions: 

60 # fall back to the latest pre-release 

61 

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

63 if len(not_yanked) > 0: 

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

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

66 return version 

67 

68 # there are no non-yanked versions: fatal 

69 raise NoVersionsError() 

70 

71 

72def download_crate(name: str, version: Optional[str] = None) -> tuple[str, str]: 

73 if version is None: 

74 # query crates.io for the latest version 

75 version = query_newest_version(name) 

76 

77 os.makedirs(CACHE_DIR, exist_ok=True) 

78 

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

80 crate_file_path = os.path.join(CACHE_DIR, crate_file_name) 

81 

82 if not os.path.isfile(crate_file_path): 

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

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

85 req.raise_for_status() 

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

87 

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

89 for chunk in tqdm.tqdm( 

90 req.iter_content(), 

91 f"Downloading {crate_file_name}", 

92 total=total, 

93 unit="B", 

94 unit_scale=True, 

95 ): 

96 f.write(chunk) 

97 

98 return crate_file_path, version