Coverage for webapp/packages/logic.py: 49%
147 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-27 22:07 +0000
1import re
3import yaml
5from flask import make_response
6from typing import List, Dict, TypedDict, Any, Union
8from canonicalwebteam.exceptions import StoreApiError
9from webapp.observability.utils import trace_function
10from webapp.store.logic import format_slug
11from webapp.store_api import publisher_gateway
14Packages = TypedDict(
15 "Packages",
16 {
17 "packages": List[
18 Dict[
19 str,
20 Union[Dict[str, Union[str, List[str]]], List[Dict[str, str]]],
21 ]
22 ]
23 },
24)
26Package = TypedDict(
27 "Package",
28 {
29 "package": Dict[
30 str, Union[Dict[str, str], List[str], List[Dict[str, str]]]
31 ]
32 },
33)
36@trace_function
37def get_icon(media):
38 icons = [m["url"] for m in media if m["type"] == "icon"]
39 if len(icons) > 0:
40 return icons[0]
41 return ""
44@trace_function
45def fetch_packages(fields: List[str], query_params) -> Packages:
46 """
47 Fetches packages from the store API based on the specified fields.
49 :param: fields (List[str]): A list of fields to include in the package
50 data.
51 :param: query_params: A search query
53 :returns: a dictionary containing the list of fetched packages.
54 """
56 category = query_params.get("categories", "")
57 query = query_params.get("q", "")
58 package_type = query_params.get("type", None)
59 platform = query_params.get("platforms", "")
60 architecture = query_params.get("architecture", "")
61 provides = query_params.get("provides", None)
62 requires = query_params.get("requires", None)
64 if package_type == "all":
65 package_type = None
67 args = {
68 "category": category,
69 "fields": fields,
70 "query": query,
71 }
73 if package_type:
74 args["type"] = package_type
76 if provides:
77 provides = provides.split(",")
78 args["provides"] = provides
80 if requires:
81 requires = requires.split(",")
82 args["requires"] = requires
84 packages = publisher_gateway.find(**args).get("results", [])
86 if platform and platform != "all":
87 filtered_packages = []
88 for p in packages:
89 platforms = p["result"].get("deployable-on", [])
90 if not platforms:
91 platforms = ["vm"]
92 if platform in platforms:
93 filtered_packages.append(p)
94 packages = filtered_packages
96 if architecture and architecture != "all":
97 args["architecture"] = architecture
98 packages = publisher_gateway.find(**args).get("results", [])
100 return packages
103@trace_function
104def fetch_package(package_name: str, fields: List[str]) -> Package:
105 """
106 Fetches a package from the store API based on the specified package name.
108 :param: package_name (str): The name of the package to fetch.
109 :param: fields (List[str]): A list of fields to include in the package
111 :returns: a dictionary containing the fetched package.
112 """
113 package = publisher_gateway.get_item_details(
114 name=package_name,
115 fields=fields,
116 api_version=2,
117 )
118 response = make_response({"package": package})
119 response.cache_control.max_age = 3600
120 return response.json
123@trace_function
124def get_bundle_charms(charm_apps):
125 result = []
127 if charm_apps:
128 for app_name, data in charm_apps.items():
129 # Charm names could be with the old prefix/suffix
130 # Like: cs:~charmed-osm/mariadb-k8s-35
131 name = data["charm"]
132 if name.startswith("cs:") or name.startswith("ch:"):
133 name = re.match(r"(?:cs:|ch:)(?:.+/)?(\S*?)(?:-\d+)?$", name)[
134 1
135 ]
137 charm = {"display_name": format_slug(name), "name": name}
139 result.append(charm)
141 return result
144@trace_function
145def parse_package_for_card(
146 package: Dict[str, Any],
147 libraries: bool = False,
148) -> Package:
149 """
150 Parses a package (charm, or bundle) and returns the formatted package
151 based on the given card schema.
153 :param: package (Dict[str, Any]): The package to be parsed.
154 :returns: a dictionary containing the formatted package.
156 note:
157 - This function has to be refactored to be more generic,
158 so we won't have to check for the package type before parsing.
160 """
161 resp = {
162 "package": {
163 "description": "",
164 "display_name": "",
165 "icon_url": "",
166 "name": "",
167 "platforms": [],
168 "type": "",
169 "channel": {
170 "name": "",
171 "risk": "",
172 "track": "",
173 },
174 },
175 "publisher": {"display_name": "", "name": "", "validation": ""},
176 "categories": [],
177 # hardcoded temporarily until we have this data from the API
178 "ratings": {"value": "0", "count": "0"},
179 }
181 result = package.get("result", {})
182 publisher = result.get("publisher", {})
183 channel = package.get("default-release", {}).get("channel", {})
184 risk = channel.get("risk", "")
185 track = channel.get("track", "")
186 if libraries:
187 resp["package"]["libraries"] = publisher_gateway.get_charm_libraries(
188 package["name"]
189 ).get("libraries", [])
190 resp["package"]["type"] = package.get("type", "")
191 resp["package"]["name"] = package.get("name", "")
192 resp["package"]["description"] = result.get("summary", "")
193 resp["package"]["display_name"] = result.get(
194 "title", format_slug(package.get("name", ""))
195 )
196 resp["package"]["channel"]["risk"] = risk
197 resp["package"]["channel"]["track"] = track
198 resp["package"]["channel"]["name"] = f"{track}/{risk}"
199 resp["publisher"]["display_name"] = publisher.get("display-name", "")
200 resp["publisher"]["validation"] = publisher.get("validation", "")
201 resp["categories"] = result.get("categories", [])
202 resp["package"]["icon_url"] = get_icon(result.get("media", []))
204 platforms = result.get("deployable-on", [])
205 if platforms:
206 resp["package"]["platforms"] = platforms
207 else:
208 resp["package"]["platforms"] = ["vm"]
210 if resp["package"]["type"] == "bundle":
211 name = package["name"]
212 default_release = publisher_gateway.get_item_details(
213 name, fields=["default-release"]
214 )
215 bundle_yaml = default_release["default-release"]["revision"][
216 "bundle-yaml"
217 ]
219 bundle_details = yaml.load(bundle_yaml, Loader=yaml.FullLoader)
220 bundle_charms = get_bundle_charms(
221 bundle_details.get(
222 "applications", bundle_details.get("services", [])
223 )
224 )
225 resp["package"]["charms"] = bundle_charms
227 return resp
230@trace_function
231def paginate(
232 packages: List[Packages], page: int, size: int, total_pages: int
233) -> List[Packages]:
234 """
235 Paginates a list of packages based on the specified page and size.
237 :param: packages (List[Packages]): The list of packages to paginate.
238 :param: page (int): The current page number.
239 :param: size (int): The number of packages to include in each page.
240 :param: total_pages (int): The total number of pages.
241 :returns: a list of paginated packages.
243 note:
244 - If the provided page exceeds the total number of pages, the last
245 page will be returned.
246 - If the provided page is less than 1, the first page will be returned.
247 """
249 if page > total_pages:
250 page = total_pages
251 if page < 1:
252 page = 1
254 start = (page - 1) * size
255 end = start + size
256 if end > len(packages):
257 end = len(packages)
259 return packages[start:end]
262@trace_function
263def get_packages(
264 libraries: bool,
265 fields: List[str],
266 size: int = 10,
267 query_params: Dict[str, Any] = {},
268) -> List[Dict[str, Any]]:
269 """
270 Retrieves a list of packages from the store based on the specified
271 parameters.The returned packages are paginated and parsed using the
272 card schema.
274 :param: store: The store object.
275 :param: fields (List[str]): A list of fields to include in the
276 package data.
277 :param: size (int, optional): The number of packages to include
278 in each page. Defaults to 10.
279 :param: page (int, optional): The current page number. Defaults to 1.
280 :param: query (str, optional): The search query.
281 :param: filters (Dict, optional): The filter parameters. Defaults to {}.
282 :returns: a dictionary containing the list of parsed packages and
283 the total pages
284 """
286 packages = fetch_packages(fields, query_params)
288 total_pages = -(len(packages) // -size)
290 total_pages = -(len(packages) // -size)
291 total_items = len(packages)
292 page = int(query_params.get("page", 1))
293 packages_per_page = paginate(packages, page, size, total_pages)
294 parsed_packages = []
295 for package in packages_per_page:
296 parsed_packages.append(parse_package_for_card(package, libraries))
297 res = parsed_packages
299 categories = get_store_categories()
301 return {
302 "packages": res,
303 "total_pages": total_pages,
304 "total_items": total_items,
305 "categories": categories,
306 }
309@trace_function
310def parse_categories(
311 categories_json: Dict[str, List[Dict[str, str]]],
312) -> List[Dict[str, str]]:
313 """
314 :param categories_json: The returned json from publishergw get_categories
315 :returns: A list of categories in the format:
316 [{"name": "Category", "slug": "category"}]
317 """
319 categories = []
321 if "categories" in categories_json:
322 for category in categories_json["categories"]:
323 categories.append(
324 {"slug": category, "name": format_slug(category)}
325 )
327 return categories
330@trace_function
331def get_store_categories() -> List[Dict[str, str]]:
332 """
333 Fetches all store categories.
335 :param: api_gw: The API object used to fetch the categories.
336 :returns: A list of categories in the format:
337 [{"name": "Category", "slug": "category"}]
338 """
339 try:
340 all_categories = publisher_gateway.get_categories()
341 except StoreApiError:
342 all_categories = []
344 for cat in all_categories["categories"]:
345 cat["display_name"] = format_slug(cat["name"])
347 categories = list(
348 filter(
349 lambda cat: cat["name"] != "featured", all_categories["categories"]
350 )
351 )
353 return categories
356@trace_function
357def get_package(
358 package_name: str,
359 fields: List[str],
360 libraries: bool,
361) -> Package:
362 """Get a package by name
364 :param store: The store object.
365 :param store_name: The name of the store.
366 :param package_name: The name of the package.
367 :param fields: The fields to fetch.
369 :return: A dictionary containing the package.
370 """
371 package = fetch_package(package_name, fields).get("package", {})
372 resp = parse_package_for_card(package, libraries)
373 return {"package": resp}