import os import requests from nexus.constant import NEXUS_URL, NPM_COMMON_REPO, NEXUS_USER from actions_logging.app_logging import logger from github.env import exit_on_error_and_write_summary, get_required_env_var def create_auth_and_headers(nexus_user=None, nexus_password=None): """ Create the headers and auth tuple for Nexus API calls :param nexus_user: default is None, will use the NEXUS_USER env var if not provided :param nexus_password: default is None, will use the GLOBAL_CICD_NEXUS_PASSWORD env var if not provided :return: headers, auth tuple """ if not nexus_user: nexus_user = os.getenv('NEXUS_USER', NEXUS_USER) if not nexus_password: nexus_password = get_required_env_var('GLOBAL_CICD_NEXUS_PASSWORD') headers = { "Accept": "application/json" } return headers, (nexus_user, nexus_password) def search_and_download(package_name, package_version, file_name, nexus_user=None, nexus_password=None, **kwargs): """ Search for a package in Nexus and download it to a file :param package_name: the name of the package to download from nexus :param package_version: the version of the package to download from nexus :param file_name: the name of the file to save the downloaded package :param nexus_user: default is None, will use the NEXUS_USER env var if not provided :param nexus_password: default is None, will use the GLOBAL_CICD_NEXUS_PASSWORD env var if not provided :return: True if successful, otherwise exit with error """ try: headers, nexus_auth = create_auth_and_headers(nexus_user, nexus_password) download_url = f'{NEXUS_URL}/service/rest/v1/search/assets/download' logger.info(f"Searching for {package_name} {package_version} in Nexus with url {download_url}") params = { 'name': package_name, 'version': package_version, **kwargs } response = requests.get(f'{download_url}', params=params, headers=headers, auth=nexus_auth) if response.status_code != 200: if package_version.startswith('v'): logger.warning(f"Retrying download without 'v' prefix") params['version'] = package_version[1:] response = requests.get(f'{download_url}', params=params, headers=headers, auth=nexus_auth) response.raise_for_status() with open(file_name, "wb") as f: f.write(response.content) logger.info(f"File downloaded successfully as {file_name}") except requests.exceptions.HTTPError as http_err: exit_on_error_and_write_summary(f'HTTP error occurred in search_and_download: {http_err}') except Exception as e: exit_on_error_and_write_summary(f"failed in search_and_download: {e}") def check_if_item_exists(params, nexus_user=None, nexus_password=None) -> (bool, str): """ Check if an item exists in Nexus. If it exists, return True and the item id. If it does not exist, return False and an empty string. If the response contains more than one item, return True and an empty string. :param params: dictionary of parameters to search for, e.g. {'repository': 'p81-releases', 'version': '1.1.1'} :param nexus_user: default is None, will use the NEXUS_USER env var if not provided :param nexus_password: default is None, will use the GLOBAL_CICD_NEXUS_PASSWORD env var if not provided :return: tuple of (bool, str) - (True/False, item_id) (item_id is empty string if not found) """ try: item_id = "" headers, nexus_auth = create_auth_and_headers(nexus_user, nexus_password) url = f'{NEXUS_URL}/service/rest/v1/search' # ?repository={repo}&group={group}&version={version}' search_res = requests.get(url, headers=headers, auth=nexus_auth, params=params) search_res.raise_for_status() if search_res.status_code == 200: items = search_res.json().get('items') if items: if len(items) == 1: item_id = items[0]["id"] logger.info(f"found 1 item with id: {item_id}") return True, item_id else: logger.debug(f"items in response: {items}") logger.info(f"{len(items)} items in the search results, return True and empty item_id") return True, item_id else: logger.info(f"no items found for {params}") return False, item_id else: logger.error(f"status code {search_res.status_code} received. 200 expected. return False and empty item_id") return False, item_id except requests.exceptions.HTTPError as http_err: exit_on_error_and_write_summary(f'HTTP error occurred in check_if_item_exists: {http_err}') except Exception as err: exit_on_error_and_write_summary(f'Other error occurred in check_if_item_exists: {err}') def delete_item(item_id): """ Delete an item from Nexus by item id :param item_id: for example: "bnBtLWNvbW1vbjpkNTMyOThlNTA3NWM3ZWNlNTI4ZjAxMjk1MDliYTk5Nw" :return: True if successful, otherwise exit with error """ try: headers, nexus_auth = create_auth_and_headers() url = f"{NEXUS_URL}/service/rest/v1/components/{item_id}" delete_res = requests.delete(url, headers=headers, auth=nexus_auth) delete_res.raise_for_status() if delete_res.status_code == 204: logger.info(f"item id {item_id} was successfully deleted") return True else: exit_on_error_and_write_summary(f"ERROR - failed to delete item id {item_id}") except requests.exceptions.HTTPError as http_err: exit_on_error_and_write_summary(f'HTTP error occurred in delete_item: {http_err}') except Exception as err: exit_on_error_and_write_summary(f'Other error occurred in delete_item: {err}') def get_package_versions(package_name, sort='version', direction='desc', repository=NPM_COMMON_REPO, nexus_user=None, nexus_password=None) -> list: """ Fetch versions list of a package from Nexus. """ try: headers, nexus_auth = create_auth_and_headers(nexus_user, nexus_password) params = { 'sort': sort, 'direction': direction, 'repository': repository, 'name': package_name } response = requests.get(f'{NEXUS_URL}/service/rest/v1/search/assets', params=params, headers=headers, auth=nexus_auth) response.raise_for_status() nexus_response = response.json() items = nexus_response.get('items', []) return items except requests.exceptions.HTTPError as http_err: if http_err.response.status_code == 404: logger.info_yellow(f'Nexus versions for {package_name} was not found') return [] exit_on_error_and_write_summary(f'HTTP error occurred in search_and_download: {http_err}') except Exception as e: exit_on_error_and_write_summary(f"failed in search_and_download: {e}")