diff options
Diffstat (limited to 'eos_downloader/cvp.py')
| -rw-r--r-- | eos_downloader/cvp.py | 276 |
1 files changed, 276 insertions, 0 deletions
diff --git a/eos_downloader/cvp.py b/eos_downloader/cvp.py new file mode 100644 index 0000000..6f14eb0 --- /dev/null +++ b/eos_downloader/cvp.py @@ -0,0 +1,276 @@ +#!/usr/bin/python +# coding: utf-8 -*- + +""" +CVP Uploader content +""" + +import os +from typing import List, Optional, Any +from dataclasses import dataclass +from loguru import logger +from cvprac.cvp_client import CvpClient +from cvprac.cvp_client_errors import CvpLoginError + +# from eos_downloader.tools import exc_to_str + +# logger = logging.getLogger(__name__) + + +@dataclass +class CvpAuthenticationItem: + """ + Data structure to represent Cloudvision Authentication + """ + server: str + port: int = 443 + token: Optional[str] = None + timeout: int = 1200 + validate_cert: bool = False + + +class Filer(): + # pylint: disable=too-few-public-methods + """ + Filer Helper for file management + """ + def __init__(self, path: str) -> None: + self.file_exist = False + self.filename = '' + self.absolute_path = '' + self.relative_path = path + if os.path.exists(path): + self.file_exist = True + self.filename = os.path.basename(path) + self.absolute_path = os.path.realpath(path) + + def __repr__(self) -> str: + return self.absolute_path if self.file_exist else '' + + +class CvFeatureManager(): + """ + CvFeatureManager Object to interect with Cloudvision + """ + def __init__(self, authentication: CvpAuthenticationItem) -> None: + """ + __init__ Class Creator + + Parameters + ---------- + authentication : CvpAuthenticationItem + Authentication information to use to connect to Cloudvision + """ + self._authentication = authentication + # self._cv_instance = CvpClient() + self._cv_instance = self._connect(authentication=authentication) + self._cv_images = self.__get_images() + # self._cv_bundles = self.__get_bundles() + + def _connect(self, authentication: CvpAuthenticationItem) -> CvpClient: + """ + _connect Connection management + + Parameters + ---------- + authentication : CvpAuthenticationItem + Authentication information to use to connect to Cloudvision + + Returns + ------- + CvpClient + cvprac session to cloudvision + """ + client = CvpClient() + if authentication.token is not None: + try: + client.connect( + nodes=[authentication.server], + username='', + password='', + api_token=authentication.token, + is_cvaas=True, + port=authentication.port, + cert=authentication.validate_cert, + request_timeout=authentication.timeout + ) + except CvpLoginError as error_data: + logger.error(f'Cannot connect to Cloudvision server {authentication.server}') + logger.debug(f'Error message: {error_data}') + logger.info('connected to Cloudvision server') + logger.debug(f'Connection info: {authentication}') + return client + + def __get_images(self) -> List[Any]: + """ + __get_images Collect information about images on Cloudvision + + Returns + ------- + dict + Fact returned by Cloudvision + """ + images = [] + logger.debug(' -> Collecting images') + images = self._cv_instance.api.get_images()['data'] + return images if self.__check_api_result(images) else [] + + # def __get_bundles(self): + # """ + # __get_bundles [Not In use] Collect information about bundles on Cloudvision + + # Returns + # ------- + # dict + # Fact returned by Cloudvision + # """ + # bundles = [] + # logger.debug(' -> Collecting images bundles') + # bundles = self._cv_instance.api.get_image_bundles()['data'] + # # bundles = self._cv_instance.post(url='/cvpservice/image/getImageBundles.do?queryparam=&startIndex=0&endIndex=0')['data'] + # return bundles if self.__check_api_result(bundles) else None + + def __check_api_result(self, arg0: Any) -> bool: + """ + __check_api_result Check API calls return content + + Parameters + ---------- + arg0 : any + Element to test + + Returns + ------- + bool + True if data are correct False in other cases + """ + logger.debug(arg0) + return len(arg0) > 0 + + def _does_image_exist(self, image_name: str) -> bool: + """ + _does_image_exist Check if an image is referenced in Cloudvision facts + + Parameters + ---------- + image_name : str + Name of the image to search for + + Returns + ------- + bool + True if present + """ + return any(image_name == image['name'] for image in self._cv_images) if isinstance(self._cv_images, list) else False + + def _does_bundle_exist(self, bundle_name: str) -> bool: + # pylint: disable=unused-argument + """ + _does_bundle_exist Check if an image is referenced in Cloudvision facts + + Returns + ------- + bool + True if present + """ + # return any(bundle_name == bundle['name'] for bundle in self._cv_bundles) + return False + + def upload_image(self, image_path: str) -> bool: + """ + upload_image Upload an image to Cloudvision server + + Parameters + ---------- + image_path : str + Path to the local file to upload + + Returns + ------- + bool + True if succeeds + """ + image_item = Filer(path=image_path) + if image_item.file_exist is False: + logger.error(f'File not found: {image_item.relative_path}') + return False + logger.info(f'File path for image: {image_item}') + if self._does_image_exist(image_name=image_item.filename): + logger.error("Image found in Cloudvision , Please delete it before running this script") + return False + try: + upload_result = self._cv_instance.api.add_image(filepath=image_item.absolute_path) + except Exception as e: # pylint: disable=broad-exception-caught + logger.error('An error occurred during upload, check CV connection') + logger.error(f'Exception message is: {e}') + return False + logger.debug(f'Upload Result is : {upload_result}') + return True + + def build_image_list(self, image_list: List[str]) -> List[Any]: + """ + Builds a list of the image data structures, for a given list of image names. + Parameters + ---------- + image_list : list + List of software image names + Returns + ------- + List: + Returns a list of images, with complete data or None in the event of failure + """ + internal_image_list = [] + image_data = None + success = True + + for entry in image_list: + for image in self._cv_images: + if image["imageFileName"] == entry: + image_data = image + + if image_data is not None: + internal_image_list.append(image_data) + image_data = None + else: + success = False + + return internal_image_list if success else [] + + def create_bundle(self, name: str, images_name: List[str]) -> bool: + """ + create_bundle Create a bundle with a list of images. + + Parameters + ---------- + name : str + Name of the bundle + images_name : List[str] + List of images available on Cloudvision + + Returns + ------- + bool + True if succeeds + """ + logger.debug(f'Init creation of an image bundle {name} with following images {images_name}') + all_images_present: List[bool] = [] + self._cv_images = self.__get_images() + all_images_present.extend( + self._does_image_exist(image_name=image_name) + for image_name in images_name + ) + # Bundle Create + if self._does_bundle_exist(bundle_name=name) is False: + logger.debug(f'Creating image bundle {name} with following images {images_name}') + images_data = self.build_image_list(image_list=images_name) + if images_data is not None: + logger.debug('Images information: {images_data}') + try: + data = self._cv_instance.api.save_image_bundle(name=name, images=images_data) + except Exception as e: # pylint: disable=broad-exception-caught + logger.critical(f'{e}') + else: + logger.debug(data) + return True + logger.critical('No data found for images') + return False |