# -----------------------------------------------------------------------------------------# (C) Copyright IBM Corp. 2023-2025.# https://opensource.org/licenses/BSD-3-Clause# -----------------------------------------------------------------------------------------from__future__importannotationsimportosfromtypingimportTYPE_CHECKING,castimportibm_watsonx_ai._wrappers.requestsasrequestsfromibm_watsonx_ai.metanamesimportExportMetaNamesfromibm_watsonx_ai.wml_client_errorimportWMLClientError,ApiRequestFailurefromibm_watsonx_ai.wml_resourceimportWMLResourceifTYPE_CHECKING:fromibm_watsonx_aiimportAPIClientimportpandas[docs]classExport(WMLResource):def__init__(self,client:APIClient)->None:WMLResource.__init__(self,__name__,client)self._client=clientself.ConfigurationMetaNames=ExportMetaNames()[docs]defstart(self,meta_props:dict[str,str|bool|list],space_id:str|None=None,project_id:str|None=None,)->dict:"""Start the export. You must provide the space_id or the project_id. ALL_ASSETS is by default False. You don't need to provide it unless it is set to True. You must provide one of the following in the meta_props: ALL_ASSETS, ASSET_TYPES, or ASSET_IDS. Only one of these can be provided. In the `meta_props`: ALL_ASSETS is a boolean. When set to True, it exports all assets in the given space. ASSET_IDS is an array that contains the list of assets IDs to be exported. ASSET_TYPES is used to provide the asset types to be exported. All assets of that asset type will be exported. Eg: wml_model, wml_model_definition, wml_pipeline, wml_function, wml_experiment, software_specification, hardware_specification, package_extension, script :param meta_props: metadata, to see available meta names use ``client.export_assets.ConfigurationMetaNames.get()`` :type meta_props: dict :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project: str, optional :return: Response json :rtype: dict **Example:** .. code-block:: python metadata = { client.export_assets.ConfigurationMetaNames.NAME: "export_model", client.export_assets.ConfigurationMetaNames.ASSET_IDS: ["13a53931-a8c0-4c2f-8319-c793155e7517", "13a53931-a8c0-4c2f-8319-c793155e7518"]} details = client.export_assets.start(meta_props=metadata, space_id="98a53931-a8c0-4c2f-8319-c793155e4598") .. code-block:: python metadata = { client.export_assets.ConfigurationMetaNames.NAME: "export_model", client.export_assets.ConfigurationMetaNames.ASSET_TYPES: ["wml_model"]} details = client.export_assets.start(meta_props=metadata, space_id="98a53931-a8c0-4c2f-8319-c793155e4598") .. code-block:: python metadata = { client.export_assets.ConfigurationMetaNames.NAME: "export_model", client.export_assets.ConfigurationMetaNames.ALL_ASSETS: True} details = client.export_assets.start(meta_props=metadata, space_id="98a53931-a8c0-4c2f-8319-c793155e4598") """ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Either 'space_id' or 'project_id' has to be provided")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")Export._validate_type(meta_props,"meta_props",dict,True)self._validate_input_meta(meta_props)meta=self.ConfigurationMetaNames._generate_resource_metadata(meta_props,with_validation=True,client=self._client)start_meta={}assets={}start_meta["name"]=meta["name"]if"description"inmeta:start_meta["description"]=meta["description"]if"all_assets"notinmeta:assets.update({"all_assets":False})else:assets.update({"all_assets":meta["all_assets"]})if"asset_types"inmeta:assets.update({"asset_types":meta["asset_types"]})if"asset_ids"inmeta:assets.update(({"asset_ids":meta["asset_ids"]}))start_meta["assets"]=assetshref=self._client._href_definitions.exports_href()params:dict={}ifspace_idisnotNone:params.update({"space_id":space_id})else:params.update({"project_id":project_id})creation_response=requests.post(href,params=params,headers=self._client._get_headers(),json=start_meta)details=self._handle_response(expected_status_code=202,operationName="export start",response=creation_response,)export_id=details["metadata"]["id"]print("export job with id{} has started. Monitor status using client.export_assets.get_details api. ""Check 'help(client.export_assets.get_details)' for details on the api usage".format(export_id))returndetails def_validate_input_meta(self,meta_props:dict[str,str|bool|list])->None:if"name"notinmeta_props:raiseWMLClientError("Its mandatory to provide 'NAME' in meta_props. Example: ""client.export_assets.ConfigurationMetaNames.NAME: 'name'")if("all_assets"notinmeta_propsand"asset_ids"notinmeta_propsand"asset_types"notinmeta_props):raiseWMLClientError("Its mandatory to provide either 'ALL_ASSETS' or 'ASSET_IDS' or 'ASSET_TYPES' ""in meta_props. Example: client.export_assets.ConfigurationMetaNames.ALL_ASSETS: True")count=0if"all_assets"inmeta_props:count=count+1if"asset_ids"inmeta_props:count=count+1if"asset_types"inmeta_props:count=count+1ifcount>1:raiseWMLClientError("Only one of 'ALL_ASSETS' or 'ASSET_IDS' or 'ASSET_TYPES' can be provided")[docs]defcancel(self,export_id:str,space_id:str|None=None,project_id:str|None=None,)->str:"""Cancel an export job. `space_id` or `project_id` has to be provided. .. note:: To delete an `export_id` job, use ``delete()`` API. :param export_id: export job identifier :type export_id: str :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project_id: str, optional :returns: status ("SUCCESS" or "FAILED") :rtype: str **Example:** .. code-block:: python client.export_assets.cancel(export_id='6213cf1-252f-424b-b52d-5cdd9814956c', space_id='3421cf1-252f-424b-b52d-5cdd981495fe') """Export._validate_type(export_id,"export_id",str,True)ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Its mandatory to provide space_id or project_id")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")href=self._client._href_definitions.export_href(export_id)params:dict={}ifspace_idisnotNone:params.update({"space_id":space_id})else:params.update({"project_id":project_id})cancel_response=requests.delete(href,params=params,headers=self._client._get_headers())details=self._handle_response(expected_status_code=204,operationName="cancel export",response=cancel_response,)if"SUCCESS"==details:print("Export job cancelled")returndetails [docs]defdelete(self,export_id:str,space_id:str|None=None,project_id:str|None=None,)->str:"""Delete the given `export_id` job. `space_id` or `project_id` has to be provided. :param export_id: export job identifier :type export_id: str :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project_id: str, optional :return: status ("SUCCESS" or "FAILED") :rtype: str **Example:** .. code-block:: python client.export_assets.delete(export_id='6213cf1-252f-424b-b52d-5cdd9814956c', space_id= '98a53931-a8c0-4c2f-8319-c793155e4598') """ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Its mandatory to provide space_id or project_id")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")Export._validate_type(export_id,"export_id",str,True)href=self._client._href_definitions.export_href(export_id)params:dict={"hard_delete":True}ifspace_idisnotNone:params.update({"space_id":space_id})else:params.update({"project_id":project_id})delete_response=requests.delete(href,params=params,headers=self._client._get_headers())details=self._handle_response(expected_status_code=204,operationName="delete export job",response=delete_response,)if"SUCCESS"==details:print("Export job deleted")returndetails [docs]defget_details(self,export_id:str|None=None,space_id:str|None=None,project_id:str|None=None,limit:int|None=None,asynchronous:bool=False,get_all:bool=False,)->dict:"""Get metadata of a given export job. If no `export_id` is specified, all export metadata is returned. :param export_id: export job identifier :type export_id: str, optional :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project_id: str, optional :param limit: limit number of fetched records :type limit: int, optional :param asynchronous: if `True`, it will work as a generator :type asynchronous: bool, optional :param get_all: if `True`, it will get all entries in 'limited' chunks :type get_all: bool, optional :return: export metadata :rtype: dict (if export_id is not None) or {"resources": [dict]} (if export_id is None) **Example:** .. code-block:: python details = client.export_assets.get_details(export_id, space_id= '98a53931-a8c0-4c2f-8319-c793155e4598') details = client.export_assets.get_details() details = client.export_assets.get_details(limit=100) details = client.export_assets.get_details(limit=100, get_all=True) details = [] for entry in client.export_assets.get_details(limit=100, asynchronous=True, get_all=True): details.extend(entry) """ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Its mandatory to provide space_id or project_id")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")Export._validate_type(export_id,"export_id",str,False)Export._validate_type(limit,"limit",int,False)href=self._client._href_definitions.exports_href()params:dict={}ifspace_idisnotNone:params.update({"space_id":space_id})else:params.update({"project_id":project_id})ifexport_idisNone:returnself._get_artifact_details(href,export_id,limit,"export job",query_params=params,_async=asynchronous,_all=get_all,)else:returnself._get_artifact_details(href,export_id,limit,"export job",query_params=params) [docs]deflist(self,space_id:str|None=None,project_id:str|None=None,limit:int|None=None,)->pandas.DataFrame:"""Return export jobs in a table format. :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project_id: str, optional :param limit: limit number of fetched records :type limit: int, optional :return: pandas.DataFrame with listed connections :rtype: pandas.DataFrame **Example:** .. code-block:: python client.export_assets.list() """ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Its mandatory to provide space_id or project_id")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")get_all=self._should_get_all_values(limit)ifspace_idisnotNone:resources=self.get_details(space_id=space_id,get_all=get_all)["resources"]else:resources=self.get_details(project_id=project_id,get_all=get_all)["resources"]values=[(m["metadata"]["id"],m["metadata"]["name"],m["metadata"]["created_at"],m["entity"]["status"]["state"],)forminresources]table=self._list(values,["ID","NAME","CREATED","STATUS"],limit)returntable [docs]@staticmethoddefget_id(export_details:dict)->str:"""Get the ID of the export job from export details. :param export_details: metadata of the export job :type export_details: dict :return: ID of the export job :rtype: str **Example:** .. code-block:: python id = client.export_assets.get_id(export_details) """Export._validate_type(export_details,"export_details",object,True)returnWMLResource._get_required_element_from_dict(export_details,"export_details",["metadata","id"]) [docs]defget_exported_content(self,export_id:str,space_id:str|None=None,project_id:str|None=None,file_path:str|None=None,)->str:"""Get the exported content as a zip file. :param export_id: export job identifier :type export_id: str :param space_id: space identifier :type space_id: str, optional :param project_id: project identifier :type project_id: str, optional :param file_path: name of local file to create, this should be absolute path of the file and the file shouldn't exist :type file_path: str, optional :return: path to the downloaded function content :rtype: str **Example:** .. code-block:: python client.export_assets.get_exported_content(export_id, space_id='98a53931-a8c0-4c2f-8319-c793155e4598', file_path='/home/user/my_exported_content.zip') """ifspace_idisNoneandproject_idisNone:raiseWMLClientError("Its mandatory to provide space_id or project_id")ifspace_idisnotNoneandproject_idisnotNone:raiseWMLClientError("Either 'space_id' or 'project_id' can be provided, not both")params:dict={}ifspace_idisnotNone:params.update({"space_id":space_id})else:params.update({"project_id":project_id})Export._validate_type(file_path,"file_path",str,True)file_path=cast(str,file_path)ifos.path.isfile(file_path):raiseWMLClientError("File with name: '{}' already exists.".format(file_path))href=self._client._href_definitions.export_content_href(export_id)try:response=requests.get(href,params=params,headers=self._client._get_headers(),stream=True)ifresponse.status_code!=200:raiseApiRequestFailure("Failure during{}.".format("downloading export content"),response)downloaded_exported_content=response.contentself._logger.info("Successfully downloaded artifact with artifact_url:{}".format(href))exceptWMLClientErrorase:raiseeexceptExceptionase:raiseWMLClientError("Downloading export content with artifact_url: '{}' failed.".format(href),e,)try:withopen(file_path,"wb")asf:f.write(downloaded_exported_content)print("Successfully saved export content to file: '{}'".format(file_path))returnfile_pathexceptIOErrorase:raiseWMLClientError("Downloading export content with artifact_url: '{}' failed.".format(href),e,)