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

1# -*- coding: utf-8 -*- 

2# -- This file is part of the Apio project 

3# -- (C) 2016-2019 FPGAwars 

4# -- Author Jesús Arroyo 

5# -- License GPLv2 

6# -- Derived from: 

7# ---- Platformio project 

8# ---- (C) 2014-2016 Ivan Kravets <me@ikravets.com> 

9# ---- License Apache v2 

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

11packages release repositories. 

12""" 

13 

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

15# Motivation is simplifying the usage. 

16 

17from math import ceil 

18from pathlib import Path 

19import requests 

20from rich.progress import track 

21from apio.common.apio_console import console, fatal_error 

22 

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

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

25TIMEOUT_SECS = 30 

26 

27 

28class FileDownloader: 

29 """Class for downloading files""" 

30 

31 CHUNK_SIZE = 1024 

32 

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

34 """Initialize a FileDownloader object 

35 * INPUTs: 

36 * url: File to download (full url) 

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

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

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

40 """ 

41 

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

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

44 self._request = None 

45 

46 # -- Store the url 

47 self._url = url 

48 

49 # -- Get the file from the url 

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

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

52 

53 # -- Build the destination path 

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

55 if dest_dir: 

56 

57 # -- Add the path 

58 self.destination = dest_dir / self.fname 

59 

60 # -- Request the file 

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

62 

63 # -- Fail in case of download error. 

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

65 fatal_error( 

66 "Got an unexpected HTTP status code: " 

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

68 f"When downloading {url}", 

69 ) 

70 

71 def get_size(self) -> int: 

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

73 assert self._request is not None 

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

75 

76 def download(self): 

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

78 

79 # -- Download iterator 

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

81 

82 # -- Open destination file, for writing bytes 

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

84 

85 # -- Get the file length in Kbytes 

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

87 

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

89 for _ in track( 

90 range(num_chunks), 

91 description="Downloading", 

92 console=console(), 

93 ): 

94 

95 file.write(next(itercontent)) 

96 

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

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

99 assert next(itercontent, None) is None 

100 

101 # -- Download done! 

102 self._request.close() 

103 

104 def __del__(self): 

105 """Close any pending request""" 

106 

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

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

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

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

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

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

113 if request is not None: 

114 request.close()