Coverage for app/views/routes/packages.py: 0%

1from typing import Annotated, Literal 

2 

3from fastapi import APIRouter, BackgroundTasks, Form, Header, Request 

4from fastapi.responses import HTMLResponse 

5 

6from app.analytics.posthog import capture_package_requested_events 

7from app.chart import generate_altair_colors, generate_chart 

8from app.core.logger import logger 

9from app.core.templates import templates 

10from app.models import TimeRangeValidValues 

11from app.normalization import NormalizedPackageName 

12from app.utils import ( 

13 extract_last_script_tag, 

14 generate_hx_push_url, 

15 parse_query_params, 

16 validate_package, 

17) 

18 

19router: APIRouter = APIRouter() 

20 

21 

22@router.get("/package-list", response_class=HTMLResponse) 

23async def get_package_list( 

24 request: Request, 

25 url: Annotated[str, Header(alias="HX-Current-URL")], 

26) -> HTMLResponse: 

27 logger.info(f"Fetching package list from URL: {url}") 

28 query_params = parse_query_params(url) 

29 

30 if query_params.error: 

31 logger.warning(query_params.error) 

32 return HTMLResponse(status_code=422, content=query_params.error) 

33 

34 package_data = [] 

35 if query_params.packages: 

36 colors = generate_altair_colors(len(query_params.packages)) 

37 for package, color in zip(query_params.packages, colors): 

38 package_data.append({"name": package, "color": color}) 

39 

40 return templates.TemplateResponse( 

41 request=request, 

42 name="fragments/package_list.html", 

43 context={"package_data": package_data}, 

44 ) 

45 

46 

47@router.post("/package-list", response_class=HTMLResponse) 

48async def create_package( 

49 request: Request, 

50 package_name: Annotated[NormalizedPackageName, Form()], 

51 url: Annotated[str, Header(alias="HX-Current-URL")], 

52) -> HTMLResponse: 

53 logger.info(f"Attempting to add package: {package_name}") 

54 query_params = parse_query_params(url) 

55 

56 if query_params.error: 

57 logger.warning(query_params.error) 

58 return HTMLResponse(status_code=422, content=query_params.error) 

59 

60 if package_name in query_params.packages: 

61 logger.warning(f"Duplicate package attempt: {package_name}") 

62 return HTMLResponse( 

63 status_code=409, content=f"'{package_name}' already selected" 

64 ) 

65 

66 if not validate_package(package_name): 

67 logger.warning(f"Invalid package name: {package_name} - not found on PyPI") 

68 return HTMLResponse( 

69 status_code=422, content=f"'{package_name}' not found on PyPI" 

70 ) 

71 

72 query_params.packages.append(package_name) 

73 num_of_packages = len(query_params.packages) 

74 logger.info( 

75 f"Successfully added package: {package_name}. Total packages: {num_of_packages}" 

76 ) 

77 

78 if len(query_params.packages) == 1: 

79 query_params.show_percentage = None 

80 

81 return templates.TemplateResponse( 

82 request=request, 

83 name="fragments/package_line_item.html", 

84 context={ 

85 "package_name": package_name, 

86 "color": generate_altair_colors(num_of_packages)[-1], 

87 }, 

88 headers={"HX-Push-Url": generate_hx_push_url(query_params)}, 

89 ) 

90 

91 

92@router.delete("/package-list", response_class=HTMLResponse) 

93async def delete_package( 

94 request: Request, 

95 package_name: NormalizedPackageName, 

96 url: Annotated[str, Header(alias="HX-Current-URL")], 

97) -> HTMLResponse: 

98 logger.info(f"Attempting to delete package: {package_name}") 

99 query_params = parse_query_params(url) 

100 

101 if query_params.error: 

102 logger.warning(query_params.error) 

103 return HTMLResponse(status_code=422, content=query_params.error) 

104 

105 if package_name in query_params.packages: 

106 query_params.packages.remove(package_name) 

107 logger.info(f"Successfully removed package: {package_name}") 

108 else: 

109 logger.warning(f"Package {package_name} not found in current URL state") 

110 

111 colors = generate_altair_colors(len(query_params.packages)) 

112 

113 package_data = [ 

114 {"name": package, "color": color} 

115 for package, color in zip(query_params.packages, colors) 

116 ] 

117 

118 if len(query_params.packages) == 1: 

119 query_params.show_percentage = None 

120 

121 return templates.TemplateResponse( 

122 request=request, 

123 name="fragments/package_list.html", 

124 context={"package_data": package_data}, 

125 headers={"HX-Push-Url": generate_hx_push_url(query_params)}, 

126 ) 

127 

128 

129@router.get("/packages-graph", response_class=HTMLResponse) 

130async def get_graph( 

131 request: Request, 

132 background_tasks: BackgroundTasks, 

133 url: Annotated[str, Header(alias="HX-Current-URL")], 

134 time_range: TimeRangeValidValues | None = None, 

135 show_percentage: Literal["on"] | None = None, 

136) -> HTMLResponse: 

137 query_params = parse_query_params(url) 

138 

139 if query_params.error: 

140 logger.warning(query_params.error) 

141 return HTMLResponse(status_code=422, content=query_params.error) 

142 

143 theme = request.cookies.get("theme", "light") 

144 chart_html = "" 

145 last_script_tag = "" 

146 

147 if time_range is not None: 

148 query_params.time_range.value = time_range 

149 

150 if ( 

151 show_percentage is not None 

152 or request.headers["HX-Trigger"] == "show-percentage" 

153 ): 

154 query_params.show_percentage = show_percentage 

155 

156 if len(query_params.packages) == 1: 

157 query_params.show_percentage = None 

158 

159 background_tasks.add_task( 

160 capture_package_requested_events, 

161 packages=query_params.packages, 

162 time_range=query_params.time_range.value, 

163 show_percentage=query_params.show_percentage == "on", 

164 hx_trigger=request.headers.get("HX-Trigger"), 

165 request_id=request.headers.get("X-Request-ID"), 

166 ) 

167 

168 if query_params.packages: 

169 chart_html = generate_chart(query_params, theme).to_html(fullhtml=False) 

170 last_script_tag = extract_last_script_tag(chart_html) or "" 

171 

172 headers = {} 

173 

174 if request.headers["HX-Trigger"] in ["time-range", "show-percentage"]: 

175 headers["HX-Push-Url"] = generate_hx_push_url(query_params) 

176 

177 return templates.TemplateResponse( 

178 request=request, 

179 name="fragments/chart.html", 

180 context={ 

181 "chart": last_script_tag, 

182 "query_params": query_params, 

183 }, 

184 headers=headers, 

185 ) 

186 

187 

188@router.get("/embed", response_class=HTMLResponse) 

189async def get_embed( 

190 request: Request, 

191 time_range: TimeRangeValidValues | None = None, 

192 theme: Literal["light", "dark"] | None = None, 

193 show_percentage: Literal["on"] | None = None, 

194) -> HTMLResponse: 

195 """Endpoint for embedded charts that can be used in iframes.""" 

196 query_params = parse_query_params(str(request.url)) 

197 

198 if query_params.error: 

199 logger.warning(query_params.error) 

200 return HTMLResponse(status_code=422, content=query_params.error) 

201 

202 if not theme: 

203 theme = request.cookies.get("theme", "light") 

204 

205 if time_range is not None: 

206 query_params.time_range.value = time_range 

207 

208 if show_percentage is not None: 

209 query_params.show_percentage = show_percentage 

210 

211 if not query_params.packages: 

212 return HTMLResponse(content="No packages selected") 

213 

214 chart_html = generate_chart(query_params, theme).to_html(fullhtml=True) 

215 

216 attribution_style = """ 

217 <style> 

218 .pypacktrends-attribution { 

219 text-align: center; 

220 padding: 8px; 

221 font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; 

222 font-size: 12px; 

223 } 

224 .pypacktrends-attribution a { 

225 color: #666; 

226 text-decoration: none; 

227 } 

228 .pypacktrends-attribution a:hover { 

229 text-decoration: underline; 

230 } 

231 </style> 

232 """ 

233 

234 attribution_html = f""" 

235 <div class="pypacktrends-attribution"> 

236 <a href="{str(request.url).replace("/embed", "")}" target="_blank"> 

237 View on pypacktrends.com 

238 </a> 

239 </div> 

240 """ 

241 

242 modified_html = chart_html.replace( 

243 "<body>", f"<body>{attribution_style}{attribution_html}" 

244 ) 

245 

246 return HTMLResponse(content=modified_html)