Coverage for webapp/store/views.py: 55%
246 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 22:43 +0000
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 22:43 +0000
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"])
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 )
112 context = {
113 "query": snap_searched,
114 "snaps": snaps_results,
115 "links": links,
116 "error_info": error_info,
117 }
119 return (
120 flask.render_template("brand-store/search.html", **context),
121 status_code,
122 )
124 @store.route("/search")
125 def store_view():
126 return flask.render_template("store/search.html")
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 = []
138 for key in allowlisted_redirect_params:
139 for value in flask.request.args.getlist(key):
140 redirect_query_items.append((key, value))
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 )
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 = []
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 = []
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 = []
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 = []
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 = []
206 categories = sorted(
207 get_categories(categories_results),
208 key=lambda category: category["slug"],
209 )
211 featured_snaps_fields = ",".join(
212 [
213 "developer_validation",
214 "media",
215 "package_name",
216 "publisher",
217 "summary",
218 "title",
219 ]
220 )
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 = {}
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 ]
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 )
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 """
261 # 404 for the snap-quarantine publisher
262 if publisher == "snap-quarantine":
263 flask.abort(404)
265 publisher_content_path = flask.current_app.config["CONTENT_DIRECTORY"][
266 "PUBLISHER_PAGES"
267 ]
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 )
275 if not context:
276 flask.abort(404)
278 popular_snaps = helpers.get_yaml(
279 publisher_content_path + publisher + "-snaps.yaml",
280 typ="safe",
281 )
283 context["popular_snaps"] = (
284 popular_snaps["snaps"] if popular_snaps else []
285 )
287 context["snaps"] = []
288 snaps_results = []
289 try:
290 snaps_results = device_gateway.find(
291 publisher=publisher,
292 fields=[
293 "title",
294 "summary",
295 "media",
296 "publisher",
297 ],
298 )["results"]
299 except StoreApiError:
300 pass # proceed with an empty list
302 for snap in snaps_results:
303 item = snap["snap"]
304 item["package_name"] = snap["name"]
305 item["icon_url"] = helpers.get_icon(item["media"])
306 context["snaps"].append(item)
308 featured_snaps = [
309 snap["package_name"] for snap in context["featured_snaps"]
310 ]
312 context["snaps"] = [
313 snap
314 for snap in context["snaps"]
315 if snap["package_name"] not in featured_snaps
316 ]
318 context["snaps_count"] = len(context["snaps"]) + len(
319 featured_snaps
320 )
322 return flask.render_template(
323 "store/publisher-details.html", **context
324 )
326 # standard page for all community publishers
327 status_code = 200
328 error_info = {}
329 snaps_results = []
330 snaps = []
331 snaps_count = 0
332 publisher_details = {"display-name": publisher, "username": publisher}
334 snaps_results = device_gateway.find(
335 publisher=publisher,
336 fields=[
337 "title",
338 "summary",
339 "media",
340 "publisher",
341 ],
342 )["results"]
344 for snap in snaps_results:
345 item = snap["snap"]
346 item["package_name"] = snap["name"]
347 item["icon_url"] = helpers.get_icon(item["media"])
348 snaps.append(item)
350 snaps_count = len(snaps)
352 if snaps_count > 0:
353 publisher_details = snaps[0]["publisher"]
355 context = {
356 "snaps": snaps,
357 "snaps_count": snaps_count,
358 "publisher": publisher_details,
359 "error_info": error_info,
360 }
362 return (
363 flask.render_template(
364 "store/community-publisher-details.html", **context
365 ),
366 status_code,
367 )
369 @store.route("/store/categories/<category>")
370 def store_category(category):
371 status_code = 200
372 error_info = {}
373 snaps_results = []
375 snaps_results = device_gateway.get_category_items(
376 category=category, size=10, page=1
377 )["results"]
378 for snap in snaps_results:
379 snap["icon_url"] = helpers.get_icon(snap["media"])
381 # if the first snap (banner snap) doesn't have an icon, remove the last
382 # snap from the list to avoid a hanging snap (grid of 9)
383 if len(snaps_results) == 10 and snaps_results[0]["icon_url"] == "":
384 snaps_results = snaps_results[:-1]
386 for index in range(len(snaps_results)):
387 snaps_results[index] = logic.get_snap_banner_url(
388 snaps_results[index]
389 )
391 context = {
392 "category": category,
393 "has_featured": True,
394 "snaps": snaps_results,
395 "error_info": error_info,
396 }
398 return (
399 flask.render_template("store/_category-partial.html", **context),
400 status_code,
401 )
403 @store.route("/store/featured-snaps/<category>")
404 def featured_snaps_in_category(category):
405 snaps_results = []
407 snaps_results = device_gateway.get_category_items(
408 category=category, size=3, page=1
409 )["_embedded"]["clickindex:package"]
411 for snap in snaps_results:
412 snap["icon_url"] = helpers.get_icon(snap["media"])
414 return flask.jsonify(snaps_results)
416 @store.route("/store/sitemap.xml")
417 def sitemap():
418 sitemap = redis_cache.get("sitemap:xml", expected_type=str)
419 if sitemap:
420 response = flask.make_response(sitemap)
421 response.headers["Content-Type"] = "application/xml"
422 response.headers["Cache-Control"] = "public, max-age=43200"
423 return response
425 base_url = "https://snapcraft.io/store"
427 snaps = []
428 page = 0
429 url = f"https://api.snapcraft.io/api/v1/snaps/search?page={page}"
430 while url:
431 response = session.get(url)
432 try:
433 snaps_response = response.json()
434 except Exception:
435 continue
437 for snap in snaps_response["_embedded"]["clickindex:package"]:
438 try:
439 last_udpated = (
440 parser.parse(snap["last_updated"])
441 .replace(tzinfo=None)
442 .strftime("%Y-%m-%d")
443 )
444 snaps.append(
445 {
446 "url": "https://snapcraft.io/"
447 + snap["package_name"],
448 "last_udpated": last_udpated,
449 }
450 )
451 except Exception:
452 continue
453 if "next" in snaps_response["_links"]:
454 url = snaps_response["_links"]["next"]["href"]
455 else:
456 url = None
458 xml_sitemap = flask.render_template(
459 "sitemap/sitemap.xml",
460 base_url=base_url,
461 links=snaps,
462 )
464 # Cache the generated sitemap for 12 hours
465 redis_cache.set("sitemap:xml", xml_sitemap, ttl=43200)
467 response = flask.make_response(xml_sitemap)
468 response.headers["Content-Type"] = "application/xml"
469 response.headers["Cache-Control"] = "public, max-age=43200"
471 return response
473 if store_query:
474 store.add_url_rule("/", "homepage", brand_store_view)
475 store.add_url_rule("/search", "search", brand_search_snap)
476 else:
477 store.add_url_rule("/store", "homepage", store_view)
479 store.add_url_rule("/explore", "explore", explore_view)
481 @store.route("/<snap_name>/create-track", methods=["POST"])
482 @login_required
483 @csrf.exempt
484 @exchange_required
485 def post_create_track(snap_name):
486 track_name = flask.request.form["track-name"]
487 version_pattern = flask.request.form.get("version-pattern")
488 auto_phasing_percentage = flask.request.form.get(
489 "automatic-phasing-percentage"
490 )
492 if auto_phasing_percentage is not None:
493 auto_phasing_percentage = float(auto_phasing_percentage)
495 response = publisher_gateway.create_track(
496 flask.session,
497 snap_name,
498 track_name,
499 version_pattern,
500 auto_phasing_percentage,
501 )
502 if response.status_code == 201:
503 return response.json(), response.status_code
504 if response.status_code == 409:
505 return (
506 jsonify({"error": "Track already exists."}),
507 response.status_code,
508 )
509 if "error-list" in response.json():
510 return (
511 jsonify(
512 {"error": response.json()["error-list"][0]["message"]}
513 ),
514 response.status_code,
515 )
516 return response.json(), response.status_code
518 return store