# -----------------------------------------------------------------------------------------# (C) Copyright IBM Corp. 2023-2025.# https://opensource.org/licenses/BSD-3-Clause# -----------------------------------------------------------------------------------------from__future__importannotationsimportjsonimportosfromtypingimportGenerator,TYPE_CHECKINGfromrequestsimportResponsefromibm_watsonx_ai._wrappersimportrequestsfromibm_watsonx_ai.metanamesimportPkgExtnMetaNamesfromibm_watsonx_ai.utilsimportPKG_EXTN_DETAILS_TYPE,content_type_forfromibm_watsonx_ai.wml_client_errorimport(WMLClientError,ApiRequestFailure,ResourceIdByNameNotFound,)fromibm_watsonx_ai.wml_resourceimportWMLResourceifTYPE_CHECKING:frompandasimportDataFramefromibm_watsonx_aiimportAPIClient[docs]classPkgExtn(WMLResource):"""Store and manage software Packages Extension specs."""ConfigurationMetaNames=PkgExtnMetaNames()"""MetaNames for Package Extensions creation."""def__init__(self,client:APIClient):WMLResource.__init__(self,__name__,client)[docs]defget_details(self,pkg_extn_id:str)->dict:"""Get package extensions details. :param pkg_extn_id: unique ID of the package extension :type pkg_extn_id: str :return: details of the package extension :rtype: dict **Example:** .. code-block:: python pkg_extn_details = client.pkg_extn.get_details(pkg_extn_id) """PkgExtn._validate_type(pkg_extn_id,"pkg_extn_id",str,True)response=requests.get(self._client._href_definitions.get_pkg_extn_href(pkg_extn_id),params=self._client._params(),headers=self._client._get_headers(),)ifresponse.status_code==200:returnself._get_required_element_from_response(self._handle_response(200,"get hw spec details",response))else:returnself._handle_response(200,"get hw spec details",response) [docs]defstore(self,meta_props:dict,file_path:str)->dict:"""Create a package extension. :param meta_props: metadata of the package extension. To see available meta names, use: .. code-block:: python client.package_extensions.ConfigurationMetaNames.get() :type meta_props: dict :param file_path: path to the file to be uploaded as a package extension :type file_path: str :return: metadata of the package extension :rtype: dict **Example:** .. code-block:: python meta_props = { client.package_extensions.ConfigurationMetaNames.NAME: "skl_pipeline_heart_problem_prediction", client.package_extensions.ConfigurationMetaNames.DESCRIPTION: "description scikit-learn_0.20", client.package_extensions.ConfigurationMetaNames.TYPE: "conda_yml" } pkg_extn_details = client.package_extensions.store(meta_props=meta_props, file_path="/path/to/file") """# quick support for COS credentials instead of local path# TODO add error handling and cleaning (remove the file)PkgExtn._validate_type(meta_props,"meta_props",dict,True)pkg_extn_meta=self.ConfigurationMetaNames._generate_resource_metadata(meta_props,with_validation=True,client=self._client)pkg_extn_meta_json=json.dumps(pkg_extn_meta)PkgExtn._validate_type(file_path,"file_path",str,True)# Step1 : Create an assetprint("Creating package extensions")href=self._client._href_definitions.get_pkg_extns_href()creation_response=requests.post(href,params=self._client._params(),headers=self._client._get_headers(),data=pkg_extn_meta_json,)pkg_extn_details=self._handle_response(201,"creating new package_extensions",creation_response)# Step2: upload pkg extension file to presigned urlifcreation_response.status_code==201:pkg_extn_asset_id=pkg_extn_details["metadata"]["asset_id"]pkg_extn_presigned_url=pkg_extn_details["entity"]["package_extension"]["href"]FILE_SIZE_LIMIT=int(0.3*1073741824)defread_in_chunks(chunk_size:int|None=FILE_SIZE_LIMIT,)->Generator[bytes,None,None]:withopen(file_path,"rb")asfile_object:whileTrue:data=file_object.read(chunk_size)ifnotdata:breakyielddatadefsend_multipart_request(file:str,url:str)->Response:content_path=os.path.abspath(file)content_size=os.stat(content_path).st_sizeindex=0headers={}try:forchunkinread_in_chunks(FILE_SIZE_LIMIT):offset=index+len(chunk)headers["Content-Range"]="bytes%s-%s/%s"%(index,offset-1,content_size,)index=offsetresponse=requests.put(url,files={"file":(file,chunk,"application/octet-stream")},)exceptExceptionase:deletion_response=requests.delete(self._client._href_definitions.get_pkg_extn_href(pkg_extn_asset_id),params=self._client._params(),headers=self._client._get_headers(),)print(deletion_response.status_code)raiseWMLClientError("Failed while reading a file.",e)returnresponse# chunking, when finally it will work on service side. to confirm it's working, the downloaded file needs to be the same as 400MB+ upladed file## put_response = send_multipart_request(file_path,# (self._client.service_instance._wml_credentials.url if self._client.ICP_PLATFORM_SPACES else "") + pkg_extn_presigned_url)href=(self._client.credentials.urlifself._client.ICP_PLATFORM_SPACESelse"")+pkg_extn_presigned_urltry:ifos.stat(file_path).st_size==0:raiseWMLClientError("Package extension file cannot be empty")withopen(file_path,"rb")asfile_object:ifnotself._client.ICP_PLATFORM_SPACES:content_type=content_type_for(filepath=file_path)put_response=requests.put(href,data=file_object.read(),headers={"Content-Type":content_type},)else:put_response=requests.put(href,files={"file":(file_path,file_object.read(),"application/octet-stream",)},)exceptExceptionase:deletion_response=requests.delete(self._client._href_definitions.get_pkg_extn_href(pkg_extn_asset_id),params=self._client._params(),headers=self._client._get_headers(),)print(deletion_response.status_code)raiseWMLClientError("Failed while reading a file.",e)ifput_response.status_code==201orput_response.status_code==200:# Step3: Mark the upload completecomplete_response=requests.post(self._client._href_definitions.get_pkg_extn_href(pkg_extn_asset_id)+"/upload_complete",headers=self._client._get_headers(),params=self._client._params(),)ifcomplete_response.status_code==204:print("SUCCESS")returnself._get_required_element_from_response(pkg_extn_details)else:try:self.delete(pkg_extn_asset_id)except:passraiseWMLClientError("Failed while creating a package extensions "+complete_response.text)else:try:self.delete(pkg_extn_asset_id)except:passraiseWMLClientError("Failed while creating a package extensions "+put_response.text)else:raiseWMLClientError("Failed while creating a package extensions "+creation_response.text) [docs]deflist(self)->DataFrame:"""List the package extensions in a table format. :return: pandas.DataFrame with listed package extensions :rtype: pandas.DataFrame .. code-block:: python client.package_extensions.list() """href=self._client._href_definitions.get_pkg_extns_href()response=requests.get(href,params=self._client._params(),headers=self._client._get_headers())self._handle_response(200,"list pkg_extn",response)asset_details=self._handle_response(200,"list assets",response)["resources"]pkg_extn_values=[(m["metadata"]["name"],m["metadata"]["asset_id"],m["entity"]["package_extension"]["type"],m["metadata"]["created_at"],)forminasset_details]table=self._list(pkg_extn_values,["NAME","ASSET_ID","TYPE","CREATED_AT"],None,)returntable [docs]@staticmethoddefget_id(pkg_extn_details:dict)->str:"""Get the unique ID of a package extension. :param pkg_extn_details: details of the package extension :type pkg_extn_details: dict :return: unique ID of the package extension :rtype: str **Example:** .. code-block:: python asset_id = client.package_extensions.get_id(pkg_extn_details) """PkgExtn._validate_type(pkg_extn_details,"pkg_extn_details",object,True)PkgExtn._validate_type_of_details(pkg_extn_details,PKG_EXTN_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(pkg_extn_details,"pkg_extn_details",["metadata","asset_id"]) [docs]defget_id_by_name(self,pkg_extn_name:str)->str:"""Get the ID of a package extension. :param pkg_extn_name: name of the package extension :type pkg_extn_name: str :return: unique ID of the package extension :rtype: str **Example:** .. code-block:: python asset_id = client.package_extensions.get_id_by_name(pkg_extn_name) """PkgExtn._validate_type(pkg_extn_name,"pkg_extn_name",str,True)parameters=self._client._params()parameters.update(name=pkg_extn_name)response=requests.get(self._client._href_definitions.get_pkg_extns_href(),params=parameters,headers=self._client._get_headers(),)total_values=self._handle_response(200,"get pkg extn",response)["total_results"]iftotal_values!=0:pkg_extn_details=self._handle_response(200,"get pkg extn",response)["resources"]returnpkg_extn_details[0]["metadata"]["asset_id"]else:raiseResourceIdByNameNotFound(pkg_extn_name,"package extension") [docs]@staticmethoddefget_href(pkg_extn_details:dict)->str:"""Get the URL of a stored package extension. :param pkg_extn_details: details of the package extension :type pkg_extn_details: dict :return: href of the package extension :rtype: str **Example:** .. code-block:: python pkg_extn_details = client.package_extensions.get_details(pkg_extn_id) pkg_extn_href = client.package_extensions.get_href(pkg_extn_details) """PkgExtn._validate_type(pkg_extn_details,"pkg_extn_details",object,True)PkgExtn._validate_type_of_details(pkg_extn_details,PKG_EXTN_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(pkg_extn_details,"pkg_extn_details",["entity","package_extension","href"],) [docs]defdelete(self,pkg_extn_id:str)->dict|str:"""Delete a package extension. :param pkg_extn_id: unique ID of the package extension :type pkg_extn_id: str :return: status ("SUCCESS" or "FAILED") if deleted synchronously or dictionary with response :rtype: str or dict **Example:** .. code-block:: python client.package_extensions.delete(pkg_extn_id) """PkgExtn._validate_type(pkg_extn_id,"pkg_extn_id",str,True)response=requests.delete(self._client._href_definitions.get_pkg_extn_href(pkg_extn_id),params=self._client._params(),headers=self._client._get_headers(),)ifresponse.status_code==200:returnself._get_required_element_from_response(response.json())else:returnself._handle_response(204,"delete pkg extn specification",response) def_get_required_element_from_response(self,response_data:dict)->dict:WMLResource._validate_type(response_data,"pkg_extn_response",dict)ifself._client.default_space_idisnotNone:new_el={"metadata":{"space_id":response_data["metadata"]["space_id"],"name":response_data["metadata"]["name"],"asset_id":response_data["metadata"]["asset_id"],"asset_type":response_data["metadata"]["asset_type"],"created_at":response_data["metadata"]["created_at"],#'updated_at': response_data['metadata']['updated_at']},"entity":response_data["entity"],}elifself._client.default_project_idisnotNone:new_el={"metadata":{"project_id":response_data["metadata"]["project_id"],"name":response_data["metadata"]["name"],"asset_id":response_data["metadata"]["asset_id"],"asset_type":response_data["metadata"]["asset_type"],"created_at":response_data["metadata"]["created_at"],},"entity":response_data["entity"],}if"href"inresponse_data["metadata"]:href_without_host=response_data["metadata"]["href"].split(".com")[-1]new_el["metadata"].update({"href":href_without_host})returnnew_el[docs]defdownload(self,pkg_extn_id:str,filename:str)->str:"""Download a package extension. :param pkg_extn_id: unique ID of the package extension to be downloaded :type pkg_extn_id: str :param filename: filename to be used for the downloaded file :type filename: str :return: path to the downloaded package extension content :rtype: str **Example:** .. code-block:: python client.package_extensions.download(pkg_extn_id,"sample_conda.yml/custom_library.zip") """PkgExtn._validate_type(pkg_extn_id,"pkg_extn_id",str,True)pkg_extn_response=requests.get(self._client._href_definitions.get_pkg_extn_href(pkg_extn_id),params=self._client._params(),headers=self._client._get_headers(),)pkg_extn_details=self._handle_response(200,"get assets",pkg_extn_response)artifact_content_url=pkg_extn_details["entity"]["package_extension"]["href"]ifpkg_extn_response.status_code==200:ifself._client.CLOUD_PLATFORM_SPACES:att_response=requests.get(artifact_content_url)else:att_response=requests.get(self._credentials.url+artifact_content_url)ifatt_response.status_code!=200:raiseApiRequestFailure("Failure during{}.".format("downloading package extension"),att_response,)downloaded_asset=att_response.contenttry:withopen(filename,"wb")asf:f.write(downloaded_asset)print("Successfully saved package extension content to file: '{}'".format(filename))returnos.getcwd()+"/"+filenameexceptIOErrorase:raiseWMLClientError("Saving asset with artifact_url: '{}' failed.".format(filename),e,)else:raiseWMLClientError("Failed while downloading the package extension "+pkg_extn_id)