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

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 

18import requests 

19from rich.progress import track 

20from apio.utils import util 

21from apio.common.apio_console import cout, console 

22from apio.common.apio_styles import ERROR 

23 

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

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

26TIMEOUT_SECS = 30 

27 

28 

29class FileDownloader: 

30 """Class for downloading files""" 

31 

32 CHUNK_SIZE = 1024 

33 

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

35 """Initialize a FileDownloader object 

36 * INPUTs: 

37 * url: File to download (full url) 

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

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

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

41 """ 

42 

43 # -- Store the url 

44 self._url = url 

45 

46 # -- Get the file from the url 

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

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

49 

50 # -- Build the destination path 

51 self.destination = self.fname 

52 if dest_dir: 52 ↛ 58line 52 didn't jump to line 58 because the condition on line 52 was always true

53 

54 # -- Add the path 

55 self.destination = dest_dir / self.fname 

56 

57 # -- Request the file 

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

59 

60 # -- Raise an exception in case of download error... 

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

62 cout( 

63 "Got an unexpected HTTP status code: " 

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

65 f"When downloading {url}", 

66 style=ERROR, 

67 ) 

68 raise util.ApioException() 

69 

70 def get_size(self) -> int: 

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

72 

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

74 

75 def start(self): 

76 """Start the downloading of the file""" 

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 if self._request: 106 ↛ exitline 106 didn't return from function '__del__' because the condition on line 106 was always true

107 self._request.close()