Coverage for src/fjc_biographies/downloader.py: 87%

1"""Downloads the FJC's Biographical Directory of Article III Federal Judges. 

2 

3The Federal Judicial Center exports its directory every night as six CSV 

4files, one for each category of information about the judges (see 

5https://www.fjc.gov/history/judges/biographical-directory-article-iii-federal-judges-export). 

6They need no account and are a few megabytes together. They are saved under 

7their FJC names in the "raw" folder of the data folder: 

8 

9 data_folder/ 

10 └── raw/ demographics.csv, federal-judicial-service.csv, ... 

11 

12Contents: 

13 data_folder: returns the folder where the files are kept. 

14 download: downloads the FJC's files. 

15 raw_folder: returns the folder of the downloaded files. 

16 

17""" 

18 

19from __future__ import annotations 

20 

21import logging 

22import os 

23import pathlib 

24import shutil 

25import sys 

26import urllib.parse 

27import urllib.request 

28from collections.abc import Iterable 

29from typing import Any 

30 

31from . import options 

32 

33logger = logging.getLogger(__name__) 

34 

35 

36def data_folder() -> pathlib.Path: 

37 """Returns the folder where the files are kept. 

38 

39 The folder is the one named by the `FJC_BIOGRAPHIES_DATA_DIR` environment 

40 variable, if it is set. Otherwise, when fjc_biographies runs from its 

41 repository, it is `options._DATA_FOLDER`: "data/courts/fjc_biographies" 

42 two folders above the repository, beside the folder of code projects 

43 (where courtpy and fjc-idb keep their files as well). When fjc_biographies 

44 is installed as a package, it is "fjc_biographies" in the user's local 

45 application data folder on Windows ("%LOCALAPPDATA%"), or in 

46 "~/.local/share" (or "$XDG_DATA_HOME") on other systems. 

47 

48 Returns: 

49 The path of the folder. It is not created. 

50 

51 """ 

52 named = os.environ.get(options._ENV_DATA) 

53 if named: 

54 return pathlib.Path(named).expanduser() 

55 if (pathlib.Path(__file__).parents[2] / 'pyproject.toml').is_file(): 55 ↛ 57line 55 didn't jump to line 57 because the condition on line 55 was always true

56 return options._DATA_FOLDER 

57 if sys.platform == 'win32': 

58 base = ( 

59 os.environ.get('LOCALAPPDATA') 

60 or pathlib.Path.home() / 'AppData' / 'Local') 

61 return pathlib.Path(base) / 'fjc_biographies' 

62 base = ( 

63 os.environ.get('XDG_DATA_HOME') 

64 or pathlib.Path.home() / '.local' / 'share') 

65 return pathlib.Path(base) / 'fjc_biographies' 

66 

67 

68def download( 

69 categories: Iterable[str] | str | None = None, 

70 folder: pathlib.Path | str | None = None, 

71 *, 

72 overwrite: bool = False) -> dict[str, pathlib.Path]: 

73 """Downloads the FJC's files of the directory. 

74 

75 A file that was downloaded before is skipped unless `overwrite` is True. 

76 The FJC adds new judges and corrects its data as it goes, so overwrite 

77 the files to get the latest data. 

78 

79 Args: 

80 categories: the categories to download ("demographics", "service", 

81 "other_service", "education", "career", and "other_nominations"). 

82 Defaults to None, in which case every category is downloaded. 

83 folder: the data folder, whose "raw" folder the files are saved in. 

84 Defaults to None, in which case `data_folder()` is used. 

85 overwrite: whether to download a file that is already there. Defaults 

86 to False. 

87 

88 Returns: 

89 The path of each file, by its category. 

90 

91 Raises: 

92 ValueError: if a category is not one of the FJC's. 

93 

94 Examples: 

95 >>> import fjc_biographies 

96 >>> fjc_biographies.download(['service', 'demographics']) # doctest: +SKIP 

97 

98 """ 

99 chosen = _categories(categories) 

100 raw = raw_folder(folder) 

101 paths = {} 

102 for category in chosen: 

103 path = raw / options._CATEGORIES[category] 

104 if overwrite or not path.is_file(): 

105 _save(options._FJC_URL + path.name, path) 

106 else: 

107 logger.info('Skipped %s, which already exists.', path.name) 

108 paths[category] = path 

109 return paths 

110 

111 

112def raw_folder(folder: pathlib.Path | str | None = None) -> pathlib.Path: 

113 """Returns the folder of the downloaded files. 

114 

115 Args: 

116 folder: the data folder. Defaults to None, in which case 

117 `data_folder()` is used. 

118 

119 Returns: 

120 The "raw" folder of the data folder. It is not created. 

121 

122 """ 

123 root = pathlib.Path(folder).expanduser() if folder else data_folder() 

124 return root / 'raw' 

125 

126 

127""" Private Functions """ 

128 

129 

130def _categories(categories: Iterable[str] | str | None) -> list[str]: 

131 """Returns the categories that are named, in order and without repeats. 

132 

133 Args: 

134 categories: names of categories, or None for every category. 

135 

136 Returns: 

137 The names. 

138 

139 Raises: 

140 ValueError: if a name is not one of the FJC's categories. 

141 

142 """ 

143 if categories is None: 

144 return list(options._CATEGORIES) 

145 if isinstance(categories, str): 

146 categories = [categories] 

147 chosen = list(dict.fromkeys(categories)) 

148 unknown = [name for name in chosen if name not in options._CATEGORIES] 

149 if unknown: 

150 message = ( 

151 f'{unknown} are not categories of the directory, which are ' 

152 f'{list(options._CATEGORIES)}') 

153 raise ValueError(message) 

154 return chosen 

155 

156 

157def _open(url: str) -> Any: 

158 """Opens a web address with the fjc_biographies user agent. 

159 

160 Args: 

161 url: the web address. 

162 

163 Returns: 

164 The response to the request. 

165 

166 Raises: 

167 ValueError: if `url` is not an https address. 

168 

169 """ 

170 if urllib.parse.urlparse(url).scheme != 'https': 170 ↛ 173line 170 didn't jump to line 173 because the condition on line 170 was always true

171 message = f'Only https addresses can be downloaded, not {url}.' 

172 raise ValueError(message) 

173 request = urllib.request.Request( # noqa: S310 

174 url, headers = {'User-Agent': options._USER_AGENT}) 

175 return urllib.request.urlopen( # noqa: S310 

176 request, timeout = options._TIMEOUT) 

177 

178 

179def _save(url: str, path: pathlib.Path) -> pathlib.Path: 

180 """Downloads a file. 

181 

182 The file is written under a temporary name and renamed when complete, so 

183 an interrupted download does not leave a partial file behind. 

184 

185 Args: 

186 url: the address of the file. 

187 path: where to save it. Its folder is created if needed. 

188 

189 Returns: 

190 `path`. 

191 

192 """ 

193 path.parent.mkdir(parents = True, exist_ok = True) 

194 partial = path.with_name(f'{path.name}.part') 

195 logger.info('Downloading %s.', path.name) 

196 try: 

197 with _open(url) as response, partial.open('wb') as file: 

198 shutil.copyfileobj(response, file) 

199 partial.replace(path) 

200 finally: 

201 partial.unlink(missing_ok = True) 

202 return path