Coverage for webapp/store/views.py: 57%

1from math import floor

2from urllib.parse import urlencode

4import requests as api_requests

5import flask

6from dateutil import parser

7from webapp.decorators import exchange_required, login_required

8import webapp.helpers as helpers

9import webapp.store.logic as logic

10from webapp.api import requests

12from canonicalwebteam.exceptions import StoreApiError

13from canonicalwebteam.store_api.dashboard import Dashboard

14from canonicalwebteam.store_api.publishergw import PublisherGW

15from canonicalwebteam.store_api.devicegw import DeviceGW

16from canonicalwebteam.snap_recommendations import SnapRecommendations

18from webapp.api.exceptions import ApiError

19from webapp.store.snap_details_views import snap_details_views

20from webapp.helpers import api_publisher_session, api_session

21from flask.json import jsonify

22from webapp.extensions import csrf

23from webapp.store.logic import (

24 get_categories,

25)

26from cache.cache_utility import redis_cache

28session = requests.Session()

30dashboard = Dashboard(api_session)

31publisher_gateway = PublisherGW("snap", api_publisher_session)

32device_gateway = DeviceGW("snap", api_session)

33snap_recommendations = SnapRecommendations(session)

36def store_blueprint(store_query=None):

37 store = flask.Blueprint(

38 "store",

39 __name__,

40 template_folder="/templates",

41 static_folder="/static",

42 )

43 snap_details_views(store)

45 @store.route("/validation-sets", defaults={"path": ""})

46 @store.route("/validation-sets/<path:path>")

47 @login_required

48 def validation_sets(path):

49 return flask.render_template("store/publisher.html")

51 @store.route("/discover")

52 def discover():

53 return flask.redirect(flask.url_for(".homepage"))

55 def brand_store_view():

56 error_info = {}

57 status_code = 200

59 try:

60 snaps = device_gateway.get_all_items(size=16)["results"]

61 except (StoreApiError, ApiError):

62 snaps = []

64 for snap in snaps:

65 if "media" in snap:

66 snap["icon_url"] = helpers.get_icon(snap["media"])

68 return (

69 flask.render_template(

70 "brand-store/store.html", snaps=snaps, error_info=error_info

71 ),

72 status_code,

73 )

75 def brand_search_snap():

76 status_code = 200

77 snap_searched = flask.request.args.get("q", default="", type=str)

79 if not snap_searched:

80 return flask.redirect(flask.url_for(".homepage"))

82 size = flask.request.args.get("limit", default=25, type=int)

83 offset = flask.request.args.get("offset", default=0, type=int)

85 try:

86 page = floor(offset / size) + 1

87 except ZeroDivisionError:

88 size = 10

89 page = floor(offset / size) + 1

91 error_info = {}

92 searched_results = []

94 searched_results = device_gateway.search(

95 snap_searched, size=size, page=page

96 )

98 snaps_results = searched_results["results"]

100 for snap in snaps_results:

101 snap["icon_url"] = helpers.get_icon(snap["media"])

102

103 links = logic.get_pages_details(

104 flask.request.base_url,

105 (

106 searched_results["_links"]

107 if "_links" in searched_results

108 else []

109 ),

110 )

111

112 context = {

113 "query": snap_searched,

114 "snaps": snaps_results,

115 "links": links,

116 "error_info": error_info,

117 }

118

119 return (

120 flask.render_template("brand-store/search.html", **context),

121 status_code,

122 )

123

124 @store.route("/search")

125 def store_view():

126 return flask.render_template("store/search.html")

127

128 @store.route("/store")

129 def explore_view():

130 allowlisted_redirect_params = (

131 "q",

132 "categories",

133 "page",

134 "architecture",

135 )

136 redirect_query_items = []

137

138 for key in allowlisted_redirect_params:

139 for value in flask.request.args.getlist(key):

140 redirect_query_items.append((key, value))

141

142 if redirect_query_items:

143 encoded_query = urlencode(redirect_query_items, doseq=True)

144 return flask.redirect(

145 f"{flask.url_for('.store_view')}?{encoded_query}"

146 )

147

148 try:

149 popular_snaps = redis_cache.get(

150 "explore:popular-snaps", expected_type=list

151 )

152 if not popular_snaps:

153 popular_snaps = snap_recommendations.get_popular()

154 redis_cache.set(

155 "explore:popular-snaps", popular_snaps, ttl=3600

156 )

157 except api_requests.exceptions.RequestException:

158 popular_snaps = []

159

160 try:

161 recent_snaps = redis_cache.get(

162 "explore:recent-snaps", expected_type=list

163 )

164 if not recent_snaps:

165 recent_snaps = snap_recommendations.get_recent()

166 redis_cache.set("explore:recent-snaps", recent_snaps, ttl=3600)

167 except api_requests.exceptions.RequestException:

168 recent_snaps = []

169

170 try:

171 trending_snaps = redis_cache.get(

172 "explore:trending-snaps", expected_type=list

173 )

174 if not trending_snaps:

175 trending_snaps = snap_recommendations.get_trending()

176 redis_cache.set(

177 "explore:trending-snaps", trending_snaps, ttl=3600

178 )

179 except api_requests.exceptions.RequestException:

180 trending_snaps = []

181

182 try:

183 top_rated_snaps = redis_cache.get(

184 "explore:top-rated-snaps", expected_type=list

185 )

186 if not top_rated_snaps:

187 top_rated_snaps = snap_recommendations.get_top_rated()

188 redis_cache.set(

189 "explore:top-rated-snaps", top_rated_snaps, ttl=3600

190 )

191 except api_requests.exceptions.RequestException:

192 top_rated_snaps = []

193

194 try:

195 categories_results = redis_cache.get(

196 "explore:categories", expected_type=list

197 )

198 if not categories_results:

199 categories_results = device_gateway.get_categories()

200 redis_cache.set(

201 "explore:categories", categories_results, ttl=3600

202 )

203 except StoreApiError:

204 categories_results = []

205

206 categories = sorted(

207 get_categories(categories_results),

208 key=lambda category: category["slug"],

209 )

210

211 featured_snaps_fields = ",".join(

212 [

213 "developer_validation",

214 "media",

215 "package_name",

216 "publisher",

217 "summary",

218 "title",

219 ]

220 )

221

222 try:

223 featured_snaps = device_gateway.get_featured_snaps(

224 fields=featured_snaps_fields

225 )

226 except (StoreApiError, api_requests.exceptions.RequestException):

227 featured_snaps = {}

228

229 currently_featured_snaps = [

230 {

231 "details": {

232 "developer_validation": snap["developer_validation"],

233 "icon": helpers.get_icon(snap["media"]),

234 "publisher": snap["publisher"],

235 "name": snap["package_name"],

236 "summary": snap["summary"],

237 "title": snap["title"],

238 }

239 }

240 for snap in featured_snaps.get("_embedded", {}).get(

241 "clickindex:package", []

242 )

243 ]

244

245 return flask.render_template(

246 "store/index.html",

247 categories=categories,

248 popular_snaps=popular_snaps,

249 recent_snaps=recent_snaps,

250 trending_snaps=trending_snaps,

251 top_rated_snaps=top_rated_snaps,

252 featured_snaps=currently_featured_snaps,

253 )

254

255 @store.route("/publisher/<regex('[a-z0-9-]*[a-z][a-z0-9-]*'):publisher>")

256 def publisher_details(publisher):

257 """

258 A view to display the publisher details page for specific publisher.

259 """

260

261 # 404 for the snap-quarantine publisher

262 if publisher == "snap-quarantine":

263 flask.abort(404)

264

265 publisher_content_path = flask.current_app.config["CONTENT_DIRECTORY"][

266 "PUBLISHER_PAGES"

267 ]

268

269 # special handling for some featured publishers with custom pages

270 if publisher in ["kde", "snapcrafters", "jetbrains"]:

271 context = helpers.get_yaml(

272 publisher_content_path + publisher + ".yaml", typ="safe"

273 )

274

275 if not context:

276 flask.abort(404)

277

278 context["snaps"] = []

279 snaps_results = logic.get_publisher_snaps(

280 device_gateway, publisher

281 )

282

283 # Map package_name to live snap data so pre-defined lists below

284 # are hydrated.

285 snaps_by_name = {}

286 for snap in snaps_results:

287 item = snap["snap"]

288 item["package_name"] = snap["name"]

289 item["icon_url"] = helpers.get_icon(item["media"])

290 snaps_by_name[snap["name"]] = item

291 context["snaps"].append(item)

292

293 # "Popular snaps" is a pre-selected but the snap data comes

294 # from the API, so unlisted/removed snaps are dropped.

295 popular_snaps = helpers.get_yaml(

296 publisher_content_path + publisher + "-snaps.yaml",

297 typ="safe",

298 )

299 context["popular_snaps"] = (

300 [

301 snaps_by_name[snap["package_name"]]

302 for snap in popular_snaps["snaps"]

303 if snap["package_name"] in snaps_by_name

304 ]

305 if popular_snaps

306 else []

307 )

308

309 context["featured_snaps"] = logic.hydrate_featured_snaps(

310 context.get("featured_snaps"), snaps_by_name

311 )

312

313 featured_snaps = [

314 snap["package_name"] for snap in context["featured_snaps"]

315 ]

316

317 context["snaps"] = [

318 snap

319 for snap in context["snaps"]

320 if snap["package_name"] not in featured_snaps

321 ]

322

323 context["snaps_count"] = len(context["snaps"]) + len(

324 featured_snaps

325 )

326

327 return flask.render_template(

328 "store/publisher-details.html", **context

329 )

330

331 # standard page for all community publishers

332 status_code = 200

333 error_info = {}

334 snaps_results = []

335 snaps = []

336 snaps_count = 0

337 publisher_details = {"display-name": publisher, "username": publisher}

338

339 snaps_results = device_gateway.find(

340 publisher=publisher,

341 fields=[

342 "title",

343 "summary",

344 "media",

345 "publisher",

346 ],

347 )["results"]

348

349 for snap in snaps_results:

350 item = snap["snap"]

351 item["package_name"] = snap["name"]

352 item["icon_url"] = helpers.get_icon(item["media"])

353 snaps.append(item)

354

355 snaps_count = len(snaps)

356

357 if snaps_count > 0:

358 publisher_details = snaps[0]["publisher"]

359

360 context = {

361 "snaps": snaps,

362 "snaps_count": snaps_count,

363 "publisher": publisher_details,

364 "error_info": error_info,

365 }

366

367 return (

368 flask.render_template(

369 "store/community-publisher-details.html", **context

370 ),

371 status_code,

372 )

373

374 @store.route("/store/categories/<category>")

375 def store_category(category):

376 status_code = 200

377 error_info = {}

378 snaps_results = []

379

380 snaps_results = device_gateway.get_category_items(

381 category=category, size=10, page=1

382 )["results"]

383 for snap in snaps_results:

384 snap["icon_url"] = helpers.get_icon(snap["media"])

385

386 # if the first snap (banner snap) doesn't have an icon, remove the last

387 # snap from the list to avoid a hanging snap (grid of 9)

388 if len(snaps_results) == 10 and snaps_results[0]["icon_url"] == "":

389 snaps_results = snaps_results[:-1]

390

391 for index in range(len(snaps_results)):

392 snaps_results[index] = logic.get_snap_banner_url(

393 snaps_results[index]

394 )

395

396 context = {

397 "category": category,

398 "has_featured": True,

399 "snaps": snaps_results,

400 "error_info": error_info,

401 }

402

403 return (

404 flask.render_template("store/_category-partial.html", **context),

405 status_code,

406 )

407

408 @store.route("/store/stats")

409 def store_stats():

410 try:

411 stats = redis_cache.get("store:stats", expected_type=dict)

412 if not stats:

413 stats = snap_recommendations.get_stats()

414 redis_cache.set("store:stats", stats, ttl=3600)

415 except (ApiError, api_requests.exceptions.RequestException):

416 return flask.jsonify({}), 503

417 return flask.jsonify(stats)

418

419 @store.route("/store/featured-snaps/<category>")

420 def featured_snaps_in_category(category):

421 snaps_results = []

422

423 snaps_results = device_gateway.get_category_items(

424 category=category, size=3, page=1

425 )["_embedded"]["clickindex:package"]

426

427 for snap in snaps_results:

428 snap["icon_url"] = helpers.get_icon(snap["media"])

429

430 return flask.jsonify(snaps_results)

431

432 @store.route("/store/sitemap.xml")

433 def sitemap():

434 sitemap = redis_cache.get("sitemap:xml", expected_type=str)

435 if sitemap:

436 response = flask.make_response(sitemap)

437 response.headers["Content-Type"] = "application/xml"

438 response.headers["Cache-Control"] = "public, max-age=43200"

439 return response

440

441 base_url = "https://snapcraft.io/store"

442

443 snaps = []

444 page = 0

445 url = f"https://api.snapcraft.io/api/v1/snaps/search?page={page}"

446 while url:

447 response = session.get(url)

448 try:

449 snaps_response = response.json()

450 except Exception:

451 continue

452

453 for snap in snaps_response["_embedded"]["clickindex:package"]:

454 try:

455 last_udpated = (

456 parser.parse(snap["last_updated"])

457 .replace(tzinfo=None)

458 .strftime("%Y-%m-%d")

459 )

460 snaps.append(

461 {

462 "url": "https://snapcraft.io/"

463 + snap["package_name"],

464 "last_udpated": last_udpated,

465 }

466 )

467 except Exception:

468 continue

469 if "next" in snaps_response["_links"]:

470 url = snaps_response["_links"]["next"]["href"]

471 else:

472 url = None

473

474 xml_sitemap = flask.render_template(

475 "sitemap/sitemap.xml",

476 base_url=base_url,

477 links=snaps,

478 )

479

480 # Cache the generated sitemap for 12 hours

481 redis_cache.set("sitemap:xml", xml_sitemap, ttl=43200)

482

483 response = flask.make_response(xml_sitemap)

484 response.headers["Content-Type"] = "application/xml"

485 response.headers["Cache-Control"] = "public, max-age=43200"

486

487 return response

488

489 if store_query:

490 store.add_url_rule("/", "homepage", brand_store_view)

491 store.add_url_rule("/search", "search", brand_search_snap)

492 else:

493 store.add_url_rule("/store", "homepage", store_view)

494

495 store.add_url_rule("/explore", "explore", explore_view)

496

497 @store.route("/<snap_name>/create-track", methods=["POST"])

498 @login_required

499 @csrf.exempt

500 @exchange_required

501 def post_create_track(snap_name):

502 track_name = flask.request.form["track-name"]

503 version_pattern = flask.request.form.get("version-pattern")

504 auto_phasing_percentage = flask.request.form.get(

505 "automatic-phasing-percentage"

506 )

507

508 if auto_phasing_percentage is not None:

509 auto_phasing_percentage = float(auto_phasing_percentage)

510

511 response = publisher_gateway.create_track(

512 flask.session,

513 snap_name,

514 track_name,

515 version_pattern,

516 auto_phasing_percentage,

517 )

518 if response.status_code == 201:

519 return response.json(), response.status_code

520 if response.status_code == 409:

521 return (

522 jsonify({"error": "Track already exists."}),

523 response.status_code,

524 )

525 if "error-list" in response.json():

526 return (

527 jsonify(

528 {"error": response.json()["error-list"][0]["message"]}

529 ),

530 response.status_code,

531 )

532 return response.json(), response.status_code

533

534 return store