# -----------------------------------------------------------------------------------------# (C) Copyright IBM Corp. 2023-2025.# https://opensource.org/licenses/BSD-3-Clause# -----------------------------------------------------------------------------------------from__future__importannotationsimportosfromtypingimportAny,TYPE_CHECKING,TypeAliasfromwarningsimportwarnimportibm_watsonx_ai._wrappers.requestsasrequestsfromibm_watsonx_ai.utilsimport(DATA_ASSETS_DETAILS_TYPE,modify_details_for_script_and_shiny,)fromibm_watsonx_ai.lifecycleimportSpecStatesfromibm_watsonx_ai.metanamesimportScriptMetaNamesfromibm_watsonx_ai.utils.utilsimport_get_id_from_deprecated_uidfromibm_watsonx_ai.wml_resourceimportWMLResourcefromibm_watsonx_ai.wml_client_errorimport(WMLClientError,ApiRequestFailure,ForbiddenActionForGitBasedProject,)ListType:TypeAlias=listifTYPE_CHECKING:fromibm_watsonx_aiimportAPIClientfrompandasimportDataFrame[docs]classScript(WMLResource):"""Store and manage script assets."""ConfigurationMetaNames=ScriptMetaNames()"""MetaNames for script assets creation."""def__init__(self,client:APIClient)->None:WMLResource.__init__(self,__name__,client)[docs]defget_details(self,script_id:str|None=None,limit:int|None=None,get_all:bool|None=None,**kwargs:Any,)->dict:"""Get script asset details. If no script_id is passed, details for all script assets are returned. :param script_id: unique ID of the script :type script_id: str, optional :param limit: limit number of fetched records :type limit: int, optional :param get_all: if True, it will get all entries in 'limited' chunks :type get_all: bool, optional :return: metadata of the stored script asset :rtype: - **dict** - if script_id is not None - **{"resources": [dict]}** - if script_id is None **Example:** .. code-block:: python script_details = client.script.get_details(script_id) """script_id=_get_id_from_deprecated_uid(kwargs,script_id,"script",can_be_none=True)defget_required_elements(response:dict[str,Any])->dict[str,Any]:response=modify_details_for_script_and_shiny(response)final_response={"metadata":response["metadata"],}if"entity"inresponse:final_response["entity"]=response["entity"]try:delfinal_response["entity"]["script"]["ml_version"]exceptKeyError:passreturnfinal_responsereturnself._get_asset_based_resource(script_id,"script",get_required_elements,limit=limit,get_all=get_all) [docs]defstore(self,meta_props:dict,file_path:str)->dict:"""Create a script asset and upload content to it. :param meta_props: name to be given to the script asset :type meta_props: dict :param file_path: path to the content file to be uploaded :type file_path: str :return: metadata of the stored script asset :rtype: dict **Example:** .. code-block:: python metadata = { client.script.ConfigurationMetaNames.NAME: 'my first script', client.script.ConfigurationMetaNames.DESCRIPTION: 'description of the script', client.script.ConfigurationMetaNames.SOFTWARE_SPEC_ID: '0cdb0f1e-5376-4f4d-92dd-da3b69aa9bda' } asset_details = client.script.store(meta_props=metadata, file_path="/path/to/file") """ifself._client.project_type=="local_git_storage":raiseForbiddenActionForGitBasedProject(reason="Storing Scripts is not supported for git based project.")Script._validate_type(meta_props,"meta_props",dict,True)Script._validate_type(file_path,"file_path",str,True)script_meta=self.ConfigurationMetaNames._generate_resource_metadata(meta_props,with_validation=True,client=self._client)name,extension=os.path.splitext(file_path)response=self._create_asset(script_meta,file_path,extension)entity=response["entity"]try:delentity["script"]["ml_version"]exceptKeyError:passfinal_response={"metadata":response["metadata"],"entity":entity}returnfinal_response def_create_asset(self,script_meta:dict[str,Any],file_path:str,extension:str=".py")->dict[str,Any]:# Step1: Create a data assetname=script_meta["metadata"]["name"]ifscript_meta["metadata"].get("description")isnotNone:desc=script_meta["metadata"]["description"]else:desc=""ifscript_meta.get("software_spec_uid")isnotNone:ifscript_meta["software_spec_uid"]=="":raiseWMLClientError("Failed while creating a script asset, SOFTWARE_SPEC_ID cannot be empty")ifextension==".py":lang="python3"elifextension==".R":lang="R"else:raiseWMLClientError("This file type is not supported. It has to be either a python script(.py file ) or a ""R script")# check if the software spec specified is base or derived and accordingly update the entity for asset creation# everything inside "if" should work.sw_spec_details=self._client.software_specifications.get_details(script_meta["software_spec_uid"])iflang=="R":rscript_sw_specs=[]# names of supported r script software specificationsretired_rscript_sw_specs=([])# names of retired r script software specificationsconstricted_rscript_sw_specs=([])# names of constricted r script software specificationsdeprecated_rscript_sw_specs=([])# names of deprecated r script software specificationsifself._client.CPD_version>=4.7:forsw_specinself._client.software_specifications.get_details()["resources"]:if"life_cycle"insw_spec["metadata"]:sw_configuration=(sw_spec["entity"].get("software_specification",{}).get("software_configuration",{}))if(sw_configuration.get("platform",{}).get("name")=="r"andlen(sw_configuration.get("included_packages",[]))>1):if(SpecStates.RETIRED.valueinsw_spec["metadata"]["life_cycle"]):retired_rscript_sw_specs.append(sw_spec["metadata"]["name"])elif(SpecStates.CONSTRICTED.valueinsw_spec["metadata"]["life_cycle"]):constricted_rscript_sw_specs.append(sw_spec["metadata"]["name"])elif(SpecStates.DEPRECATED.valueinsw_spec["metadata"]["life_cycle"]):deprecated_rscript_sw_specs.append(sw_spec["metadata"]["name"])elif"rstudio"notinsw_spec["metadata"]["name"].lower():rscript_sw_specs.append(sw_spec["metadata"]["name"])elifself._client.CPD_version==4.6:rscript_sw_specs=["runtime-22.2-r4.2"]deprecated_rscript_sw_specs=["default_r3.6","runtime-22.1-r3.6"]else:rscript_sw_specs=["default_r3.6","runtime-22.1-r3.6"]deprecated_rscript_sw_specs=[]rscript_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameinrscript_sw_specs]retired_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameinretired_rscript_sw_specs]constricted_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameinconstricted_rscript_sw_specs]deprecated_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameindeprecated_rscript_sw_specs]unsupported_dict={SpecStates.RETIRED.value:retired_sw_spec_ids,SpecStates.CONSTRICTED.value:constricted_sw_spec_ids,SpecStates.DEPRECATED.value:deprecated_sw_spec_ids,}if(script_meta["software_spec_uid"]notinrscript_sw_spec_ids+deprecated_sw_spec_ids+constricted_sw_spec_ids+retired_sw_spec_ids):raiseWMLClientError(f"For R scripts, only base software specs{', '.join(rscript_sw_specs)} ""are supported. Specify ""the id you get via ""client.software_specifications.get_id_by_name(sw_name)")warning_error_msg=("Provided software spec is{key} for R scripts. "+f"Only base software specs{', '.join(rscript_sw_specs)} are supported."f" Specify the id you get via "f"client.software_specifications.get_id_by_name(sw_name)")forkeyinunsupported_dict:ifscript_meta["software_spec_uid"]inunsupported_dict[key]:matchkey:caseSpecStates.RETIRED.value:raiseWMLClientError(warning_error_msg.format(key=key))case_:software_spec_warning=warning_error_msg.format(key=key)warn(software_spec_warning,category=RuntimeWarning)ifsw_spec_details["entity"]["software_specification"]["type"]=="base":asset_meta={"metadata":{"name":name,"description":desc,"asset_type":"script","origin_country":"us","asset_category":"USER",},"entity":{"script":{"ml_version":"4.0.0","language":{"name":lang},"software_spec":{"base_id":script_meta["software_spec_uid"]},}},}elifsw_spec_details["entity"]["software_specification"]["type"]=="derived":asset_meta={"metadata":{"name":name,"description":desc,"asset_type":"script","origin_country":"us","asset_category":"USER",},"entity":{"script":{"ml_version":"4.0.0","language":{"name":lang},"software_spec":{"id":script_meta["software_spec_uid"]},}},}else:asset_meta={"metadata":{"name":name,"description":desc,"asset_type":"script","origin_country":"us","asset_category":"USER",},"entity":{"script":{"ml_version":"4.0.0","language":{"name":lang},"software_spec":{"id":script_meta["software_spec_uid"]},}},}# Step1: Create an assetprint("Creating Script asset...")ifself._client.CLOUD_PLATFORM_SPACES:creation_response=requests.post(self._client._href_definitions.get_assets_href(),headers=self._client._get_headers(),params=self._client._params(),json=asset_meta,)else:# if self._client.ICP_PLATFORM_SPACEScreation_response=requests.post(self._client._href_definitions.get_data_assets_href(),headers=self._client._get_headers(),params=self._client._params(),json=asset_meta,)asset_details=self._handle_response(201,"creating new asset",creation_response)# Step2: Create attachmentifcreation_response.status_code==201:asset_id=asset_details["metadata"]["asset_id"]attachment_meta={"asset_type":"script","name":"attachment_"+asset_id}attachment_response=requests.post(self._client._href_definitions.get_attachments_href(asset_id),headers=self._client._get_headers(),params=self._client._params(),json=attachment_meta,)attachment_details=self._handle_response(201,"creating new attachment",attachment_response)ifattachment_response.status_code==201:attachment_id=attachment_details["attachment_id"]attachment_url=attachment_details["url1"]# Step3: Put content to attachmenttry:withopen(file_path,"rb")asf:ifnotself._client.ICP_PLATFORM_SPACES:put_response=requests.put(attachment_url,data=f.read())else:put_response=requests.put(self._credentials.url+attachment_url,files={"file":(name,f,"application/octet-stream")},)exceptExceptionase:deletion_response=requests.delete(self._client._href_definitions.get_data_asset_href(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:# Step4: Complete attachmentcomplete_response=requests.post(self._client._href_definitions.get_attachment_complete_href(asset_id,attachment_id),headers=self._client._get_headers(),params=self._client._params(),)ifcomplete_response.status_code==200:print("SUCCESS")returnself._get_required_element_from_response(asset_details)else:try:self.delete(asset_id)except:passraiseWMLClientError("Failed while creating a script asset. Try again.")else:try:self.delete(asset_id)except:passraiseWMLClientError("Failed while creating a script asset. Try again.")else:print("SUCCESS")returnself._get_required_element_from_response(asset_details)else:raiseWMLClientError("Failed while creating a script asset. Try again.")[docs]deflist(self,limit:int|None=None)->DataFrame:"""List stored scripts in a table format. :param limit: limit number of fetched records :type limit: int, optional :return: pandas.DataFrame with listed scripts :rtype: pandas.DataFrame **Example:** .. code-block:: python client.script.list() """Script._validate_type(limit,"limit",int,False)href=self._client._href_definitions.get_search_script_href()data:dict[str,Any]={"query":"*:*"}iflimitisnotNone:data.update({"limit":limit})response=requests.post(href,params=self._client._params(),headers=self._client._get_headers(),json=data,)self._handle_response(200,"list assets",response)asset_details=self._handle_response(200,"list assets",response)["results"]space_values=[(m["metadata"]["name"],m["metadata"]["asset_type"],m["metadata"]["asset_id"],)forminasset_details]table=self._list(space_values,["NAME","ASSET_TYPE","ASSET_ID"],None)returntable [docs]defdownload(self,asset_id:str|None=None,filename:str|None=None,rev_id:str|None=None,**kwargs:Any,)->str:"""Download the content of a script asset. :param asset_id: unique ID of the script asset to be downloaded :type asset_id: str :param filename: filename to be used for the downloaded file :type filename: str :param rev_id: revision ID :type rev_id: str, optional :return: path to the downloaded asset content :rtype: str **Example:** .. code-block:: python client.script.download(asset_id, "script_file") """iffilenameisNone:raiseTypeError("download() missing 1 required positional argument: 'filename'")asset_id=_get_id_from_deprecated_uid(kwargs,asset_id,"asset",can_be_none=False)rev_id=_get_id_from_deprecated_uid(kwargs,rev_id,"rev",can_be_none=True)# Backward compatibility in past `rev_id` was an int.ifisinstance(rev_id,int):rev_id_as_int_warning="`rev_id` parameter type as int is deprecated, please convert to str instead"warn(rev_id_as_int_warning,category=DeprecationWarning)rev_id=str(rev_id)Script._validate_type(asset_id,"asset_id",str,True)Script._validate_type(rev_id,"rev_id",str,False)params=self._client._params()ifrev_idisnotNone:params.update({"revision_id":rev_id})importurllibifnotself._client.ICP_PLATFORM_SPACES:asset_response=requests.get(self._client._href_definitions.get_asset_href(asset_id),params=params,headers=self._client._get_headers(),)else:asset_response=requests.get(self._client._href_definitions.get_data_asset_href(asset_id),params=params,headers=self._client._get_headers(),)asset_details=self._handle_response(200,"get assets",asset_response)attachment_id=asset_details["attachments"][0]["id"]response=requests.get(self._client._href_definitions.get_attachment_href(asset_id,attachment_id),params=params,headers=self._client._get_headers(),)ifresponse.status_code==200:attachment_signed_url=response.json()["url"]if"connection_id"inasset_details["attachments"][0]:att_response=requests.get(attachment_signed_url)else:ifnotself._client.ICP_PLATFORM_SPACES:att_response=requests.get(attachment_signed_url)else:att_response=requests.get(self._credentials.url+attachment_signed_url)ifatt_response.status_code!=200:raiseApiRequestFailure("Failure during{}.".format("downloading asset"),att_response)downloaded_asset=att_response.contenttry:withopen(filename,"wb")asf:f.write(downloaded_asset)print("Successfully saved data asset content to file: '{}'".format(filename))returnos.path.abspath(filename)exceptIOErrorase:raiseWMLClientError("Saving asset with artifact_url to local file: '{}' failed.".format(filename),e,)else:raiseWMLClientError("Failed while downloading the asset "+asset_id)# type: ignore [docs]@staticmethoddefget_id(asset_details:dict)->str:"""Get the unique ID of a stored script asset. :param asset_details: metadata of the stored script asset :type asset_details: dict :return: unique ID of the stored script asset :rtype: str **Example:** .. code-block:: python asset_id = client.script.get_id(asset_details) """Script._validate_type(asset_details,"asset_details",object,True)Script._validate_type_of_details(asset_details,DATA_ASSETS_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(asset_details,"data_assets_details",["metadata","guid"]) [docs]@staticmethoddefget_href(asset_details:dict)->str:"""Get the URL of a stored script asset. :param asset_details: details of the stored script asset :type asset_details: dict :return: href of the stored script asset :rtype: str **Example:** .. code-block:: python asset_details = client.script.get_details(asset_id) asset_href = client.script.get_href(asset_details) """Script._validate_type(asset_details,"asset_details",object,True)Script._validate_type_of_details(asset_details,DATA_ASSETS_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(asset_details,"asset_details",["metadata","href"]) [docs]defupdate(self,script_id:str|None=None,meta_props:dict|None=None,file_path:str|None=None,**kwargs:Any,)->dict:"""Update a script with metadata, attachment, or both. :param script_id: ID of the script :type script_id: str :param meta_props: changes for the script matadata :type meta_props: dict, optional :param file_path: file path to the new attachment :type file_path: str, optional :return: updated metadata of the script :rtype: dict **Example:** .. code-block:: python script_details = client.script.update(script_id, meta, content_path) """script_id=_get_id_from_deprecated_uid(kwargs,script_id,"script",can_be_none=False)Script._validate_type(script_id,"script_id",str,True)ifmeta_propsisNoneandfile_pathisNone:raiseWMLClientError("At least either meta_props or file_path has to be provided")updated_details=Noneurl=self._client._href_definitions.get_asset_href(script_id)# STEPS# STEP 1. Get existing metadata# STEP 2. If meta_props provided, we need to patch meta# a. Construct meta patch string and call /v2/assets/<asset_id> to patch meta# b. Construct entity patch if required and call /v2/assets/<asset_id>/attributes/script to patch entity# STEP 3. If file_path provided, we need to patch the attachment# a. If attachment already exists for the script, delete it# b. POST call to get signed url for upload# c. Upload to the signed url# d. Mark upload complete# STEP 4. Get the updated script record and return# STEP 1response=requests.get(url,params=self._client._params(),headers=self._client._get_headers())ifresponse.status_code!=200:ifresponse.status_code==404:raiseWMLClientError("Invalid input. Unable to get the details of script_id provided.")else:raiseApiRequestFailure("Failure during{}.".format("getting script to update"),response)details=self._handle_response(200,"Get script details",response)attachments_response=None# STEP 2a.# Patch meta if providedifmeta_propsisnotNone:self._validate_type(meta_props,"meta_props",dict,True)meta_patch_payload=[]entity_patch_payload=[]# Since we are dealing with direct asset apis, there can be metadata or entity patch or bothif"name"inmeta_propsor"description"inmeta_props:props_for_asset_meta_patch={}forkeyinmeta_props:ifkey=="name"orkey=="description":props_for_asset_meta_patch.update({key:meta_props[key]})meta_patch_payload=(self.ConfigurationMetaNames._generate_patch_payload(details,props_for_asset_meta_patch,with_validation=True))# STEP 2b.if"software_spec_uid"inmeta_props:ifdetails["entity"]["script"]["software_spec"]:entity_patch_payload=[{"op":"replace","path":"/software_spec/base_id","value":meta_props["software_spec_uid"],}]else:entity_patch_payload=[{"op":"add","path":"/software_spec","value":"{base_id:"+meta_props["software_spec_uid"]+"}",}]ifmeta_patch_payload:meta_patch_url=self._client._href_definitions.get_asset_href(script_id)response_patch=requests.patch(meta_patch_url,json=meta_patch_payload,params=self._client._params(),headers=self._client._get_headers(),)updated_details=self._handle_response(200,"script patch",response_patch)ifentity_patch_payload:entity_patch_url=(self._client._href_definitions.get_asset_href(script_id)+"/attributes/script")response_patch=requests.patch(entity_patch_url,json=entity_patch_payload,params=self._client._params(),headers=self._client._get_headers(),)updated_details=self._handle_response(200,"script patch",response_patch)iffile_pathisnotNone:if"attachments"indetailsanddetails["attachments"]:current_attachment_id=details["attachments"][0]["id"]else:current_attachment_id=None# STEP 3attachments_response=self._update_attachment_for_assets("script",script_id,file_path,current_attachment_id)ifattachments_responseisnotNoneand"success"notinattachments_response:self._update_msg(updated_details)# Have to fetch again to reflect updated asset and attachment idsurl=self._client._href_definitions.get_asset_href(script_id)response=requests.get(url,params=self._client._params(),headers=self._client._get_headers())ifresponse.status_code!=200:ifresponse.status_code==404:raiseWMLClientError("Invalid input. Unable to get the details of script_id provided.")else:raiseApiRequestFailure("Failure during{}.".format("getting script to update"),response)response=self._get_required_element_from_response(self._handle_response(200,"Get script details",response))entity=response["entity"]try:delentity["script"]["ml_version"]exceptKeyError:passfinal_response={"metadata":response["metadata"],"entity":entity}returnfinal_response def_update_msg(self,updated_details:Any)->None:ifupdated_detailsisnotNone:print("Could not update the attachment because of server error."" However metadata is updated. Try updating attachment again later")else:raiseWMLClientError("Unable to update attachment because of server error. Try again later")[docs]defdelete(self,asset_id:str|None=None,**kwargs:Any)->dict|str:"""Delete a stored script asset. :param asset_id: ID of the script asset :type asset_id: str :return: status ("SUCCESS" or "FAILED") if deleted synchronously or dictionary with response :rtype: str | dict **Example:** .. code-block:: python client.script.delete(asset_id) """asset_id=_get_id_from_deprecated_uid(kwargs,asset_id,"asset",can_be_none=False)Script._validate_type(asset_id,"asset_id",str,True)ifself._if_deployment_exist_for_asset(asset_id):raiseWMLClientError("Cannot delete script that has existing deployments. Please delete all associated deployments and try again")response=requests.delete(self._client._href_definitions.get_asset_href(asset_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 assets",response) [docs]defcreate_revision(self,script_id:str|None=None,**kwargs:Any)->dict:"""Create a revision for the given script. Revisions are immutable once created. The metadata and attachment at `script_id` is taken and a revision is created out of it. :param script_id: ID of the script :type script_id: str :return: revised metadata of the stored script :rtype: dict **Example:** .. code-block:: python script_revision = client.script.create_revision(script_id) """script_id=_get_id_from_deprecated_uid(kwargs,script_id,"script",can_be_none=False)# For CP4D, check if either space or project ID is setself._client._check_if_either_is_set()Script._validate_type(script_id,"script_id",str,True)print("Creating script revision...")response=self._get_required_element_from_response(self._create_revision_artifact_for_assets(script_id,"Script"))entity=response["entity"]try:delentity["script"]["ml_version"]exceptKeyError:passfinal_response={"metadata":response["metadata"],"entity":entity}returnfinal_response [docs]deflist_revisions(self,script_id:str|None=None,limit:int|None=None,**kwargs:Any)->DataFrame:"""Print all revisions for the given script ID in a table format. :param script_id: ID of the stored script :type script_id: str :param limit: limit number of fetched records :type limit: int, optional :return: pandas.DataFrame with listed revisions :rtype: pandas.DataFrame **Example:** .. code-block:: python client.script.list_revisions(script_id) """script_id=_get_id_from_deprecated_uid(kwargs,script_id,"script",can_be_none=False)# For CP4D, check if either space or project ID is setself._client._check_if_either_is_set()Script._validate_type(script_id,"script_id",str,True)url=self._client._href_definitions.get_asset_href(script_id)+"/revisions"# /v2/assets/{asset_id}/revisions returns 'results' objectscript_resources=self._get_with_or_without_limit(url,None,"List Script revisions",summary=None,pre_defined=None,_all=self._should_get_all_values(limit),)["resources"]script_values=[(m["metadata"]["asset_id"],m["metadata"]["revision_id"],m["metadata"]["name"],m["metadata"]["commit_info"]["committed_at"],)forminscript_resources]table=self._list(script_values,["ID","REVISION_ID","NAME","REVISION_COMMIT"],limit,)returntable [docs]defget_revision_details(self,script_id:str|None=None,rev_id:str|None=None,**kwargs:Any)->ListType:"""Get metadata of the script revision. :param script_id: ID of the script :type script_id: str :param rev_id: ID of the revision. If this parameter is not provided, it returns the latest revision. If there is no latest revision, it returns an error. :type rev_id: str, optional :return: metadata of the stored script(s) :rtype: list **Example:** .. code-block:: python script_details = client.script.get_revision_details(script_id, rev_id) """script_id=_get_id_from_deprecated_uid(kwargs,script_id,"script",can_be_none=False)rev_id=_get_id_from_deprecated_uid(kwargs,rev_id,"rev",can_be_none=True)# Backward compatibility in past `rev_id` was an int.ifisinstance(rev_id,int):rev_id_as_int_warning="`rev_id` parameter type as int is deprecated, please convert to str instead"warn(rev_id_as_int_warning,category=DeprecationWarning)rev_id=str(rev_id)# For CP4D, check if either space or project ID is setself._client._check_if_either_is_set()Script._validate_type(script_id,"script_id",str,True)Script._validate_type(rev_id,"rev_id",str,False)ifrev_idisNone:rev_id="latest"url=self._client._href_definitions.get_asset_href(script_id)resources=self._get_with_or_without_limit(url,limit=None,op_name="asset_revision",summary=None,pre_defined=None,revision=rev_id,)["resources"]responses=[self._get_required_element_from_response(resource)forresourceinresources]final_responses=[]forresponseinresponses:entity=response["entity"]try:delentity["script"]["ml_version"]exceptKeyError:passfinal_responses.append({"metadata":response["metadata"],"entity":entity})returnfinal_responses def_get_required_element_from_response(self,response_data:dict)->dict:WMLResource._validate_type(response_data,"scripts",dict)revision_id=Nonemetadata={"name":response_data["metadata"]["name"],"guid":response_data["metadata"]["asset_id"],"href":response_data["href"],"asset_type":response_data["metadata"]["asset_type"],"created_at":response_data["metadata"]["created_at"],"last_updated_at":response_data["metadata"]["usage"]["last_updated_at"],}try:ifself._client.default_space_idisnotNone:metadata["space_id"]=response_data["metadata"]["space_id"]elifself._client.default_project_idisnotNone:metadata["project_id"]=response_data["metadata"]["project_id"]if"description"inresponse_data["metadata"]:metadata.update({"description":response_data["metadata"]["description"]})if"revision_id"inresponse_data["metadata"]:revision_id=response_data["metadata"]["revision_id"]metadata.update({"revision_id":response_data["metadata"]["revision_id"]})if"attachments"inresponse_dataandresponse_data["attachments"]:metadata.update({"attachment_id":response_data["attachments"][0]["id"]})if"commit_info"inresponse_data["metadata"]andrevision_idisnotNone:metadata.update({"revision_commit_date":response_data["metadata"]["commit_info"]["committed_at"]})new_el={"metadata":metadata,"entity":response_data["entity"]}ifself._client.default_project_idisnotNone:href_without_host=response_data["href"].split(".com")[-1]new_el["metadata"].update({"href":href_without_host})returnnew_elexceptException:raiseWMLClientError(f"Failed to read Response from down-stream service:{response_data}")