Coverage for webapp/packages/logic.py: 15%
131 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-28 22:05 +0000
1import datetime
3from flask import make_response
4from typing import List, Dict, TypedDict, Any, Union
6from canonicalwebteam.exceptions import StoreApiError
7from canonicalwebteam.store_api.devicegw import DeviceGW
9from webapp.helpers import get_icon
10from webapp.helpers import api_session
13device_gateway = DeviceGW("snap", api_session)
15Packages = TypedDict(
16 "Packages",
17 {
18 "packages": List[
19 Dict[
20 str,
21 Union[Dict[str, Union[str, List[str]]], List[Dict[str, str]]],
22 ]
23 ]
24 },
25)
27Package = TypedDict(
28 "Package",
29 {
30 "package": Dict[
31 str, Union[Dict[str, str], List[str], List[Dict[str, str]]]
32 ]
33 },
34)
37def fetch_packages(fields: List[str], query_params) -> Packages:
38 """
39 Fetches packages from the store API based on the specified fields.
41 :param: fields (List[str]): A list of fields to include in the package
42 data.
43 :param: query_params: A search query
45 :returns: a dictionary containing the list of fetched packages.
46 """
48 category = query_params.get("categories", "")
49 query = query_params.get("q", "")
50 package_type = query_params.get("type", None)
51 platform = query_params.get("platforms", "")
52 architecture = query_params.get("architecture", "")
53 provides = query_params.get("provides", None)
54 requires = query_params.get("requires", None)
56 if package_type == "all":
57 package_type = None
59 args = {
60 "category": category,
61 "fields": fields,
62 "query": query,
63 }
65 if package_type:
66 args["type"] = package_type
68 if provides:
69 provides = provides.split(",")
70 args["provides"] = provides
72 if requires:
73 requires = requires.split(",")
74 args["requires"] = requires
76 packages = device_gateway.find(**args).get("results", [])
78 if platform and platform != "all":
79 filtered_packages = []
80 for p in packages:
81 platforms = p["result"].get("deployable-on", [])
82 if not platforms:
83 platforms = ["vm"]
84 if platform in platforms:
85 filtered_packages.append(p)
86 packages = filtered_packages
88 if architecture and architecture != "all":
89 args["architecture"] = architecture
90 packages = device_gateway.find(**args).get("results", [])
92 return packages
95def fetch_package(package_name: str, fields: List[str]) -> Package:
96 """
97 Fetches a package from the store API based on the specified package name.
99 :param: package_name (str): The name of the package to fetch.
100 :param: fields (List[str]): A list of fields to include in the package
102 :returns: a dictionary containing the fetched package.
103 """
104 package = device_gateway.get_item_details(
105 name=package_name,
106 fields=fields,
107 api_version=2,
108 )
109 response = make_response({"package": package})
110 response.cache_control.max_age = 3600
111 return response.json
114def parse_package_for_card(package: Dict[str, Any]) -> Package:
115 """
116 Parses a snap and returns the formatted package
117 based on the given card schema.
119 :param: package (Dict[str, Any]): The package to be parsed.
120 :returns: a dictionary containing the formatted package.
122 note:
123 - This function has to be refactored to be more generic,
124 so we won't have to check for the package type before parsing.
126 """
127 resp = {
128 "package": {
129 "description": "",
130 "display_name": "",
131 "icon_url": "",
132 "name": "",
133 "platforms": [],
134 "type": "",
135 "channel": {
136 "name": "",
137 "risk": "",
138 "track": "",
139 },
140 },
141 "publisher": {"display_name": "", "name": "", "validation": ""},
142 "categories": [],
143 # hardcoded temporarily until we have this data from the API
144 "ratings": {"value": "0", "count": "0"},
145 }
147 snap = package.get("snap", {})
148 publisher = snap.get("publisher", {})
149 resp["package"]["description"] = snap.get("summary", "")
150 resp["package"]["display_name"] = snap.get("title", "")
151 resp["package"]["type"] = "snap"
152 resp["package"]["name"] = package.get("name", "")
153 # platform to be fetched
154 resp["publisher"]["display_name"] = publisher.get("display-name", "")
155 resp["publisher"]["name"] = publisher.get("username", "")
156 resp["publisher"]["validation"] = publisher.get("validation", "")
157 resp["categories"] = snap.get("categories", [])
158 resp["package"]["icon_url"] = get_icon(package["snap"]["media"])
160 return resp
163def paginate(
164 packages: List[Packages], page: int, size: int, total_pages: int
165) -> List[Packages]:
166 """
167 Paginates a list of packages based on the specified page and size.
169 :param: packages (List[Packages]): The list of packages to paginate.
170 :param: page (int): The current page number.
171 :param: size (int): The number of packages to include in each page.
172 :param: total_pages (int): The total number of pages.
173 :returns: a list of paginated packages.
175 note:
176 - If the provided page exceeds the total number of pages, the last
177 page will be returned.
178 - If the provided page is less than 1, the first page will be returned.
179 """
181 if page > total_pages:
182 page = total_pages
183 if page < 1:
184 page = 1
186 start = (page - 1) * size
187 end = start + size
188 if end > len(packages):
189 end = len(packages)
191 return packages[start:end]
194def get_packages(
195 fields: List[str],
196 size: int = 10,
197 query_params: Dict[str, Any] = {},
198) -> List[Dict[str, Any]]:
199 """
200 Retrieves a list of packages from the store based on the specified
201 parameters.The returned packages are paginated and parsed using the
202 card schema.
204 :param: store: The store object.
205 :param: fields (List[str]): A list of fields to include in the
206 package data.
207 :param: size (int, optional): The number of packages to include
208 in each page. Defaults to 10.
209 :param: page (int, optional): The current page number. Defaults to 1.
210 :param: query (str, optional): The search query.
211 :param: filters (Dict, optional): The filter parameters. Defaults to {}.
212 :returns: a dictionary containing the list of parsed packages and
213 the total pages
214 """
216 packages = fetch_packages(fields, query_params)
218 total_pages = -(len(packages) // -size)
220 total_pages = -(len(packages) // -size)
221 total_items = len(packages)
222 page = int(query_params.get("page", 1))
223 packages_per_page = paginate(packages, page, size, total_pages)
224 parsed_packages = []
225 for package in packages_per_page:
226 parsed_packages.append(parse_package_for_card(package))
227 res = parsed_packages
229 categories = get_store_categories()
231 return {
232 "packages": res,
233 "total_pages": total_pages,
234 "total_items": total_items,
235 "categories": categories,
236 }
239def format_slug(slug):
240 """Format category name into a standard title format
242 :param slug: The hypen spaced, lowercase slug to be formatted
243 :return: The formatted string
244 """
245 return (
246 slug.title()
247 .replace("-", " ")
248 .replace("And", "and")
249 .replace("Iot", "IoT")
250 )
253def parse_categories(
254 categories_json: Dict[str, List[Dict[str, str]]]
255) -> List[Dict[str, str]]:
256 """
257 :param categories_json: The returned json from store_api.get_categories()
258 :returns: A list of categories in the format:
259 [{"name": "Category", "slug": "category"}]
260 """
262 categories = []
264 if "categories" in categories_json:
265 for category in categories_json["categories"]:
266 categories.append(
267 {"slug": category, "name": format_slug(category)}
268 )
270 return categories
273def get_store_categories() -> List[Dict[str, str]]:
274 """
275 Fetches all store categories.
277 :returns: A list of categories in the format:
278 [{"name": "Category", "slug": "category"}]
279 """
280 try:
281 all_categories = device_gateway.get_categories()
282 except StoreApiError:
283 all_categories = []
285 for cat in all_categories["categories"]:
286 cat["display_name"] = format_slug(cat["name"])
288 categories = list(
289 filter(
290 lambda cat: cat["name"] != "featured", all_categories["categories"]
291 )
292 )
294 return categories
297def get_snaps_account_info(account_info):
298 """Get snaps from the account information of a user
300 :param account_info: The account informations
302 :return: A list of snaps
303 :return: A list of registred snaps
304 """
305 user_snaps = {}
306 registered_snaps = {}
307 if "16" in account_info["snaps"]:
308 snaps = account_info["snaps"]["16"]
309 for snap in snaps.keys():
310 if snaps[snap]["status"] != "Revoked":
311 if not snaps[snap]["latest_revisions"]:
312 registered_snaps[snap] = snaps[snap]
313 else:
314 user_snaps[snap] = snaps[snap]
316 now = datetime.datetime.utcnow()
318 for snap in user_snaps:
319 snap_info = user_snaps[snap]
320 for revision in snap_info["latest_revisions"]:
321 if len(revision["channels"]) > 0:
322 snap_info["latest_release"] = revision
323 break
325 if len(user_snaps) == 1:
326 for snap in user_snaps:
327 snap_info = user_snaps[snap]
328 revisions = snap_info["latest_revisions"]
330 revision_since = datetime.datetime.strptime(
331 revisions[-1]["since"], "%Y-%m-%dT%H:%M:%SZ"
332 )
334 if abs((revision_since - now).days) < 30 and (
335 not revisions[0]["channels"]
336 or revisions[0]["channels"][0] == "edge"
337 ):
338 snap_info["is_new"] = True
340 return user_snaps, registered_snaps
343def get_package(
344 package_name: str,
345 fields: List[str],
346) -> Package:
347 """Get a package by name
349 :param store: The store object.
350 :param store_name: The name of the store.
351 :param package_name: The name of the package.
352 :param fields: The fields to fetch.
354 :return: A dictionary containing the package.
355 """
356 package = fetch_package(package_name, fields).get("package", {})
357 resp = parse_package_for_card(package)
358 return {"package": resp}