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 

18from pathlib import Path 

19import requests 

20from rich.progress import track 

21from apio.utils import util 

22from apio.common.apio_console import cout, console 

23from apio.common.apio_styles import ERROR 

24 

25 

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

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

28TIMEOUT_SECS = 30 

29 

30 

31class FileDownloader: 

32 """Class for downloading files""" 

33 

34 CHUNK_SIZE = 1024 

35 

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

37 """Initialize a FileDownloader object 

38 * INPUTs: 

39 * url: File to download (full url) 

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

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

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

43 """ 

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: 54 ↛ 60line 54 didn't jump to line 60 because the condition on line 54 was always true

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 # -- Raise an exception 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 cout( 

65 "Got an unexpected HTTP status code: " 

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

67 f"When downloading {url}", 

68 style=ERROR, 

69 ) 

70 raise util.ApioException() 

71 

72 def get_size(self) -> int: 

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

74 

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

76 

77 def start(self): 

78 """Start the downloading of the file""" 

79 

80 # -- Download iterator 

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

82 

83 # -- Open destination file, for writing bytes 

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

85 

86 # -- Get the file length in Kbytes 

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

88 

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

90 for _ in track( 

91 range(num_chunks), 

92 description="Downloading", 

93 console=console(), 

94 ): 

95 

96 file.write(next(itercontent)) 

97 

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

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

100 assert next(itercontent, None) is None 

101 

102 # -- Download done! 

103 self._request.close() 

104 

105 def __del__(self): 

106 """Close any pending request""" 

107 

108 if self._request: 108 ↛ exitline 108 didn't return from function '__del__' because the condition on line 108 was always true

109 self._request.close()