Coverage for webapp/packages/logic.py: 21%
144 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-18 22:11 +0000
1import re
3import yaml
5from flask import make_response
6from typing import List, Dict, TypedDict, Any, Union
8from canonicalwebteam.exceptions import StoreApiError
9from redis_cache.cache_utility import redis_cache
10from webapp.observability.utils import trace_function
11from webapp.store.logic import format_slug
12from webapp.store_api import publisher_gateway
13from webapp.config import CATEGORIES
16Packages = TypedDict(
17 "Packages",
18 {
19 "packages": List[
20 Dict[
21 str,
22 Union[Dict[str, Union[str, List[str]]], List[Dict[str, str]]],
23 ]
24 ]
25 },
26)
28Package = TypedDict(
29 "Package",
30 {
31 "package": Dict[
32 str, Union[Dict[str, str], List[str], List[Dict[str, str]]]
33 ]
34 },
35)
38@trace_function
39def get_icon(media):
40 icons = [m["url"] for m in media if m["type"] == "icon"]
41 if len(icons) > 0:
42 return icons[0]
43 return ""
46@trace_function
47def fetch_packages(
48 fields: List[str], query_params: Dict[str, Any]
49) -> list[Package]:
50 """
51 Fetches and parses packages from the store API.
53 :param: fields (List[str]): A list of fields to include in the package
54 data.
55 :param: query_params: A search query
57 :returns: a list of parsed packages.
58 """
60 category = query_params.get("categories", "")
61 query = query_params.get("q", "")
62 package_type = query_params.get("type", None)
63 platform = query_params.get("platforms", "")
64 provides = query_params.get("provides", None)
65 requires = query_params.get("requires", None)
67 args = {
68 "category": category,
69 "fields": fields,
70 "query": query,
71 }
73 if provides:
74 args["provides"] = provides.split(",")
76 if requires:
77 args["requires"] = requires.split(",")
79 if package_type and package_type != "all":
80 args["type"] = package_type
82 key = (
83 "fetch-packages",
84 {
85 **query_params,
86 "fields": tuple(fields),
87 },
88 )
89 result = redis_cache.get(key, expected_type=list)
90 if result:
91 return result
92 packages = publisher_gateway.find(**args).get("results", [])
93 if platform and platform != "all":
94 filtered_packages = []
95 for p in packages:
96 platforms = p.get("result", {}).get("deployable-on", [])
97 if not platforms:
98 platforms = ["vm"]
99 if platform in platforms:
100 filtered_packages.append(p)
101 packages = filtered_packages
103 result = [parse_package_for_card(package) for package in packages]
104 redis_cache.set(key, result, ttl=600)
106 return result
109@trace_function
110def fetch_package(package_name: str, fields: List[str]) -> Package:
111 """
112 Fetches a package from the store API based on the specified package name.
114 :param: package_name (str): The name of the package to fetch.
115 :param: fields (List[str]): A list of fields to include in the package
117 :returns: a dictionary containing the fetched package.
118 """
119 package = publisher_gateway.get_item_details(
120 name=package_name,
121 fields=fields,
122 api_version=2,
123 )
124 response = make_response({"package": package})
125 response.cache_control.max_age = 3600
126 return response.json
129@trace_function
130def get_bundle_charms(charm_apps):
131 result = []
133 if charm_apps:
134 for _, data in charm_apps.items():
135 # Charm names could be with the old prefix/suffix
136 # Like: cs:~charmed-osm/mariadb-k8s-35
137 name = data["charm"]
138 if name.startswith("cs:") or name.startswith("ch:"):
139 name = re.match(r"(?:cs:|ch:)(?:.+/)?(\S*?)(?:-\d+)?$", name)[
140 1
141 ]
143 charm = {"display_name": format_slug(name), "name": name}
145 result.append(charm)
147 return result
150@trace_function
151def parse_package_for_card(
152 package: Dict[str, Any],
153 libraries: bool = False,
154) -> Package:
155 """
156 Parses a package (charm, or bundle) and returns the formatted package
157 based on the given card schema.
159 :param: package (Dict[str, Any]): The package to be parsed.
160 :returns: a dictionary containing the formatted package.
162 note:
163 - This function has to be refactored to be more generic,
164 so we won't have to check for the package type before parsing.
166 """
167 resp = {
168 "package": {
169 "description": "",
170 "display_name": "",
171 "icon_url": "",
172 "name": "",
173 "platforms": [],
174 "type": "",
175 "channel": {
176 "name": "",
177 "risk": "",
178 "track": "",
179 },
180 },
181 "publisher": {"display_name": "", "name": "", "validation": ""},
182 "categories": [],
183 # hardcoded temporarily until we have this data from the API
184 "ratings": {"value": "0", "count": "0"},
185 }
187 result = package.get("result", {})
188 publisher = result.get("publisher", {})
189 channel = package.get("default-release", {}).get("channel", {})
190 risk = channel.get("risk", "")
191 track = channel.get("track", "")
192 if libraries:
193 resp["package"]["libraries"] = publisher_gateway.get_charm_libraries(
194 package["name"]
195 ).get("libraries", [])
196 resp["package"]["type"] = package.get("type", "")
197 resp["package"]["name"] = package.get("name", "")
198 resp["package"]["description"] = result.get("summary", "")
199 resp["package"]["display_name"] = result.get(
200 "title", format_slug(package.get("name", ""))
201 )
202 resp["package"]["channel"]["risk"] = risk
203 resp["package"]["channel"]["track"] = track
204 resp["package"]["channel"]["name"] = f"{track}/{risk}"
205 resp["publisher"]["display_name"] = publisher.get("display-name", "")
206 resp["publisher"]["validation"] = publisher.get("validation", "")
207 resp["categories"] = result.get("categories", [])
208 resp["package"]["icon_url"] = get_icon(result.get("media", []))
210 platforms = result.get("deployable-on", [])
211 if platforms:
212 resp["package"]["platforms"] = platforms
213 else:
214 resp["package"]["platforms"] = ["vm"]
216 if resp["package"]["type"] == "bundle":
217 name = package["name"]
218 default_release = publisher_gateway.get_item_details(
219 name, fields=["default-release"]
220 )
221 bundle_yaml = default_release["default-release"]["revision"][
222 "bundle-yaml"
223 ]
225 bundle_details = yaml.load(bundle_yaml, Loader=yaml.FullLoader)
226 bundle_charms = get_bundle_charms(
227 bundle_details.get(
228 "applications", bundle_details.get("services", [])
229 )
230 )
231 resp["package"]["charms"] = bundle_charms
233 return resp
236@trace_function
237def paginate(
238 packages: List[Packages], page: int, size: int, total_pages: int
239) -> List[Packages]:
240 """
241 Paginates a list of packages based on the specified page and size.
243 :param: packages (List[Packages]): The list of packages to paginate.
244 :param: page (int): The current page number.
245 :param: size (int): The number of packages to include in each page.
246 :param: total_pages (int): The total number of pages.
247 :returns: a list of paginated packages.
249 note:
250 - If the provided page exceeds the total number of pages, the last
251 page will be returned.
252 - If the provided page is less than 1, the first page will be returned.
253 """
255 if page > total_pages:
256 page = total_pages
257 if page < 1:
258 page = 1
260 start = (page - 1) * size
261 end = start + size
262 if end > len(packages):
263 end = len(packages)
265 return packages[start:end]
268@trace_function
269def get_packages(
270 fields: List[str],
271 query_params: Dict[str, Any],
272 size: int = 10,
273) -> Dict[str, Any]:
274 """
275 Retrieves a list of packages and paginate.
277 :param: fields (List[str]): A list of fields to include in the
278 package data.
279 :param: size (int, optional): The number of packages to include
280 in each page. Defaults to 10.
281 :param: page (int, optional): The current page number. Defaults to 1.
282 :param: query (str, optional): The search query.
283 :param: filters (Dict, optional): The filter parameters. Defaults to {}.
284 :returns: a dictionary containing the list of parsed packages and
285 the total pages
286 """
288 packages = fetch_packages(fields, query_params)
290 total_pages = -(len(packages) // -size)
291 total_items = len(packages)
292 page = int(query_params.get("page", 1))
294 res = paginate(packages, page, size, total_pages)
296 categories = get_store_categories()
298 return {
299 "packages": res,
300 "total_pages": total_pages,
301 "total_items": total_items,
302 "categories": categories,
303 }
306@trace_function
307def get_store_categories() -> List[Dict[str, str]]:
308 """
309 Fetches all store categories.
311 :param: api_gw: The API object used to fetch the categories.
312 :returns: A list of categories in the format:
313 [{"name": "Category", "slug": "category"}]
314 """
315 key = "store-categories"
316 categories = redis_cache.get(key, expected_type=list)
317 if not categories:
318 try:
319 all_categories = publisher_gateway.get_categories()
320 except StoreApiError:
321 all_categories = []
323 category_map = {cat["slug"]: cat["name"] for cat in CATEGORIES}
325 for cat in all_categories["categories"]:
326 cat["display_name"] = category_map.get(
327 cat["name"], format_slug(cat["name"])
328 )
330 categories = list(
331 filter(
332 lambda cat: cat["name"] != "featured",
333 all_categories["categories"],
334 )
335 )
336 redis_cache.set(key, categories, ttl=3600)
337 return categories
340@trace_function
341def get_package(
342 package_name: str,
343 fields: List[str],
344 libraries: bool,
345) -> Package:
346 """Get a package by name
348 :param store: The store object.
349 :param store_name: The name of the store.
350 :param package_name: The name of the package.
351 :param fields: The fields to fetch.
353 :return: A dictionary containing the package.
354 """
356 key = (
357 f"get-package:{package_name}:lib-{libraries}"
358 if libraries
359 else f"get-package:{package_name}"
360 )
361 resp = redis_cache.get(key, expected_type=dict)
362 if not resp:
363 package = fetch_package(package_name, fields).get("package", {})
364 resp = parse_package_for_card(package, libraries)
365 redis_cache.set(key, resp, ttl=600)
367 return {"package": resp}