Coverage for apio/managers/downloader.py: 95%

1# This file is part of the Apio project 

2# (C) 2016-present FPGAwars 

3# Project authors: 

4# * Jesús Arroyo (Jesus89) (2016-2019) 

5# * Juan Gonzalez (obijuan) (2019-2024) 

6# * The Apio Team (2024-present) 

7# License: GPLv2 

8 

9"""Implement a remote file downloader. Used to fetch packages from github 

10packages release repositories. 

11""" 

12 

13# TODO: capture all the exceptions and return them as method return status. 

14# Motivation is simplifying the usage. 

15 

16from math import ceil 

17from pathlib import Path 

18import requests 

19from rich.progress import track 

20from apio.common.apio_console import console, fatal_error 

21 

22# -- Timeout for getting a response from the server when downloading 

23# -- a file (in seconds). We had github tests failing with timeout=10 

24TIMEOUT_SECS = 30 

25 

26 

27class FileDownloader: 

28 """Class for downloading files""" 

29 

30 CHUNK_SIZE = 1024 

31 

32 def __init__(self, url: str, dest_dir=None): 

33 """Initialize a FileDownloader object 

34 * INPUTs: 

35 * url: File to download (full url) 

36 (Ex. 'https://github.com/FPGAwars/apio-examples/ 

37 releases/download/0.0.35/apio-examples-0.0.35.zip') 

38 * dest_dir: Destination folder (where to download the file) 

39 """ 

40 

41 # -- Initialize the request field first, so that __del__ works even 

42 # -- if the construction fails below, e.g. on a connection timeout. 

43 self._request = None 

44 

45 # -- Store the url 

46 self._url = url 

47 

48 # -- Get the file from the url 

49 # -- Ex: 'apio-examples-0.0.35.zip' 

50 self.fname = url.split("/")[-1] 

51 

52 # -- Build the destination path 

53 self.destination: Path = Path(self.fname) 

54 if dest_dir: 

55 

56 # -- Add the path 

57 self.destination = dest_dir / self.fname 

58 

59 # -- Request the file 

60 self._request = requests.get(url, stream=True, timeout=TIMEOUT_SECS) 

61 

62 # -- Fail in case of download error. 

63 if self._request.status_code != 200: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 fatal_error( 

65 "Got an unexpected HTTP status code: " 

66 f"{self._request.status_code}", 

67 f"When downloading {url}", 

68 ) 

69 

70 def get_size(self) -> int: 

71 """Return the size (in bytes) of the latest bytes block received""" 

72 assert self._request is not None 

73 return int(self._request.headers["content-length"]) 

74 

75 def download(self): 

76 """Download of the file while displaying a progress bar.""" 

77 

78 # -- Download iterator 

79 itercontent = self._request.iter_content(chunk_size=self.CHUNK_SIZE) 

80 

81 # -- Open destination file, for writing bytes 

82 with open(self.destination, "wb") as file: 

83 

84 # -- Get the file length in Kbytes 

85 num_chunks = int(ceil(self.get_size() / float(self.CHUNK_SIZE))) 

86 

87 # -- Download and write the chunks, while displaying the progress. 

88 for _ in track( 

89 range(num_chunks), 

90 description="Downloading", 

91 console=console(), 

92 ): 

93 

94 file.write(next(itercontent)) 

95 

96 # -- Check that the iterator reached its end. When the end is 

97 # -- reached, next() returns the default value None. 

98 assert next(itercontent, None) is None 

99 

100 # -- Download done! 

101 self._request.close() 

102 

103 def __del__(self): 

104 """Close any pending request""" 

105 

106 # -- Using getattr() since __del__ can run on a partially constructed 

107 # -- object, and an exception here would surface as an 'unraisable 

108 # -- exception' at an arbitrary point of the program. Comparing to 

109 # -- None rather than testing truthiness since bool(Response) is 

110 # -- Response.ok, which is False for non-2xx error responses. 

111 request = getattr(self, "_request", None) 

112 if request is not None: 

113 request.close()