Movatterモバイル変換

[docs]classShiny(WMLResource):"""Store and manage shiny assets."""ConfigurationMetaNames=ShinyMetaNames()"""MetaNames for Shiny Assets creation."""def__init__(self,client:APIClient):WMLResource.__init__(self,__name__,client)

[docs]defget_details(self,shiny_id:str|None=None,limit:int|None=None,get_all:bool|None=None,**kwargs:Any,)->dict[str,Any]:"""Get shiny asset details. If no shiny_id is passed, details for all shiny assets are returned. :param shiny_id: unique ID of the shiny asset :type shiny_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 shiny asset :rtype: - **dict** - if shiny_id is not None - **{"resources": [dict]}** - if shiny_id is None **Example:** .. code-block:: python shiny_details = client.shiny.get_details(shiny_id) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",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"],"entity":{}}returnfinal_responsereturnself._get_asset_based_resource(shiny_id,"shiny_asset",get_required_elements,limit=limit,get_all=get_all)

[docs]defstore(self,meta_props:dict,file_path:str)->dict:"""Create a shiny asset and upload content to it. :param meta_props: metadata of the shiny 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 shiny asset :rtype: dict **Example:** .. code-block:: python meta_props = { client.shiny.ConfigurationMetaNames.NAME: "shiny app name" } shiny_details = client.shiny.store(meta_props, file_path="/path/to/file") """ifself._client.project_type=="local_git_storage":raiseForbiddenActionForGitBasedProject(reason="Storing Shiny apps is not supported for git based project.")Shiny._validate_type(file_path,"file_path",str,True)shiny_meta=self.ConfigurationMetaNames._generate_resource_metadata(meta_props,with_validation=True,client=self._client)response=self._create_asset(shiny_meta,file_path)final_response={"metadata":response["metadata"],"entity":{}}returnfinal_response

def_create_asset(self,shiny_meta:dict[str,Any],file_path:str)->dict[str,Any]:# Step1: Create a shiny assetname=shiny_meta["metadata"]["name"]desc=""ifshiny_meta["metadata"].get("description"):desc=shiny_meta["metadata"]["description"]shiny_sw_spec_id=Noneifshiny_meta.get("software_spec_uid"):shiny_sw_spec_id=shiny_meta["software_spec_uid"]shiny_sw_specs=[]deprecated_shiny_sw_specs=[]ifself._client.CPD_version>=4.8:forsw_specinself._client.software_specifications.get_details()["resources"]:ifsw_spec.get("metadata",{}).get("life_cycle",{}):if("shiny"insw_spec["metadata"]["name"]or"rstudio"insw_spec["metadata"]["name"]):if("retired"or"deprecated"or"constricted")notinsw_spec["metadata"]["life_cycle"]:shiny_sw_specs.append(sw_spec["metadata"]["name"])elif"deprecated"insw_spec["metadata"]["life_cycle"]:deprecated_shiny_sw_specs.append(sw_spec["metadata"]["name"])elifself._client.CPD_version>=4.6:shiny_sw_specs=["rstudio_r4.2"]deprecated_shiny_sw_specs=["shiny-r3.6"]else:shiny_sw_specs=["shiny-r3.6"]deprecated_shiny_sw_specs=[]shiny_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameinshiny_sw_specs]deprecated_shiny_sw_spec_ids=[self._client.software_specifications.get_id_by_name(sw_name)forsw_nameindeprecated_shiny_sw_specs]if(shiny_sw_spec_idandshiny_sw_spec_idnotinshiny_sw_spec_ids+deprecated_shiny_sw_spec_ids):raiseWMLClientError(f"For R Shiny assets, only base software specs{','.join(shiny_sw_specs)} ""are supported. Specify ""the id you get via ""self._client.software_specifications.get_id_by_name(sw_name)")elifshiny_sw_spec_idandshiny_sw_spec_idindeprecated_shiny_sw_spec_ids:software_spec_deprecated_warning=("Provided software spec is deprecated for R Shiny assets. "f"Only base software specs{','.join(shiny_sw_specs)} ""are supported. Specify the id you get via ""self._client.software_specifications.get_id_by_name(sw_name)")warn(software_spec_deprecated_warning,category=DeprecationWarning)asset_meta={"metadata":{"name":name,"description":desc,"asset_type":"shiny_asset","origin_country":"us","asset_category":"USER",},"entity":{"shiny_asset":{"ml_version":"4.0.0"}},}if(self._client.CPD_version>=4.6)andshiny_sw_spec_id:asset_meta["entity"]["shiny_asset"]["software_spec"]={"base_id":shiny_sw_spec_id}# Step1: Create an assetprint("Creating Shiny asset...")creation_response=requests.post(self._client._href_definitions.get_data_assets_href(),headers=self._client._get_headers(),params=self._client._params(),json=asset_meta,)shiny_details=self._handle_response(201,"creating new asset",creation_response)# Step2: Create attachmentifcreation_response.status_code==201:asset_id=shiny_details["metadata"]["asset_id"]attachment_meta={"asset_type":"shiny_asset","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(shiny_details)else:try:self.delete(asset_id)except:passraiseWMLClientError("Failed while creating a shiny asset. Try again.")else:try:self.delete(asset_id)except:passraiseWMLClientError("Failed while creating a shiny asset. Try again.")else:print("SUCCESS")returnself._get_required_element_from_response(shiny_details)else:raiseWMLClientError("Failed while creating a shiny asset. Try again.")

[docs]deflist(self,limit:int|None=None)->DataFrame:"""List stored shiny assets in a table format. :param limit: limit number of fetched records :type limit: int, optional :return: pandas.DataFrame with listed shiny assets :rtype: pandas.DataFrame **Example:** .. code-block:: python client.shiny.list() """Shiny._validate_type(limit,"limit",int,False)href=self._client._href_definitions.get_search_shiny_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)shiny_details=self._handle_response(200,"list assets",response)["results"]space_values=[(m["metadata"]["name"],m["metadata"]["asset_type"],m["metadata"]["asset_id"],)forminshiny_details]table=self._list(space_values,["NAME","ASSET_TYPE","ASSET_ID"],limit)returntable

[docs]defdownload(self,shiny_id:str|None=None,filename:str|None=None,rev_id:str|None=None,**kwargs:Any,)->str:"""Download the content of a shiny asset. :param shiny_id: unique ID of the shiny asset to be downloaded :type shiny_id: str :param filename: filename to be used for the downloaded file :type filename: str :param rev_id: ID of the revision :type rev_id: str, optional :return: path to the downloaded shiny asset content :rtype: str **Example:** .. code-block:: python client.shiny.download(shiny_id, "shiny_asset.zip") """iffilenameisNone:raiseTypeError("download() missing 1 required positional argument: 'filename'")shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",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)Shiny._validate_type(shiny_id,"shiny_id",str,True)Shiny._validate_type(rev_id,"rev_id",str,False)params=self._client._params()ifrev_idisnotNone:params.update({"revision_id":rev_id})asset_response=requests.get(self._client._href_definitions.get_data_asset_href(shiny_id),params=params,headers=self._client._get_headers(),)shiny_details=self._handle_response(200,"get shiny assets",asset_response)attachment_id=shiny_details["attachments"][0]["id"]response=requests.get(self._client._href_definitions.get_attachment_href(shiny_id,attachment_id),params=params,headers=self._client._get_headers(),)ifresponse.status_code==200:attachment_signed_url=response.json()["url"]if"connection_id"inshiny_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 shiny asset content to file: '{}'".format(filename))returnos.getcwd()+"/"+filenameexceptIOErrorase:raiseWMLClientError("Saving shiny asset with artifact_url to local file: '{}' failed.".format(filename),e,)else:raiseWMLClientError("Failed while downloading the shiny asset "+shiny_id)# type: ignore

[docs]@staticmethoddefget_id(shiny_details:dict)->str:"""Get the unique ID of a stored shiny asset. :param shiny_details: metadata of the stored shiny asset :type shiny_details: dict :return: unique ID of the stored shiny asset :rtype: str **Example:** .. code-block:: python shiny_id = client.shiny.get_id(shiny_details) """Shiny._validate_type(shiny_details,"shiny_details",object,True)Shiny._validate_type_of_details(shiny_details,DATA_ASSETS_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(shiny_details,"data_assets_details",["metadata","guid"])

[docs]@staticmethoddefget_uid(shiny_details:dict)->str:"""Get the Unique ID of a stored shiny asset. *Deprecated:* Use ``get_id(shiny_details)`` instead. :param shiny_details: metadata of the stored shiny asset :type shiny_details: dict :return: unique ID of the stored shiny asset :rtype: str **Example:** .. code-block:: python shiny_id = client.shiny.get_uid(shiny_details) """returnShiny.get_id(shiny_details)

[docs]@staticmethoddefget_href(shiny_details:dict)->str:"""Get the URL of a stored shiny asset. :param shiny_details: details of the stored shiny asset :type shiny_details: dict :return: href of the stored shiny asset :rtype: str **Example:** .. code-block:: python shiny_details = client.shiny.get_details(shiny_id) shiny_href = client.shiny.get_href(shiny_details) """Shiny._validate_type(shiny_details,"shiny_details",object,True)Shiny._validate_type_of_details(shiny_details,DATA_ASSETS_DETAILS_TYPE)returnWMLResource._get_required_element_from_dict(shiny_details,"shiny_details",["metadata","href"])

[docs]defupdate(self,shiny_id:str|None=None,meta_props:dict|None=None,file_path:str|None=None,**kwargs:Any,)->dict[str,Any]:"""Update a shiny asset with metadata, attachment, or both. :param shiny_id: ID of the shiny asset :type shiny_id: str :param meta_props: changes to the metadata of the shiny asset :type meta_props: dict, optional :param file_path: file path to the new attachment :type file_path: str, optional :return: updated metadata of the shiny asset :rtype: dict **Example:** .. code-block:: python shiny_details = client.shiny.update(shiny_id, meta, content_path) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",can_be_none=False)Shiny._validate_type(shiny_id,"shiny_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(shiny_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# 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 shiny_id provided.")else:raiseApiRequestFailure("Failure during{}.".format("getting shiny asset to update"),response,)details=self._handle_response(200,"Get shiny asset details",response)attachments_response=None# STEP 2a.# Patch meta if providedifmeta_propsisnotNone:self._validate_type(meta_props,"meta_props",dict,True)meta_patch_payload=[]# Since we are dealing with direct asset apis, name and description is metadata patchif"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))ifmeta_patch_payload:meta_patch_url=self._client._href_definitions.get_asset_href(shiny_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,"shiny 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("shiny_asset",shiny_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(shiny_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 shiny_id provided.")else:raiseApiRequestFailure("Failure during{}.".format("getting shiny to update"),response)response=self._get_required_element_from_response(self._handle_response(200,"Get shiny details",response))final_response={"metadata":response["metadata"],"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,shiny_id:str|None=None,**kwargs:Any)->str|dict:"""Delete a stored shiny asset. :param shiny_id: unique ID of the shiny asset :type shiny_id: str :return: status ("SUCCESS" or "FAILED") if deleted synchronously or dictionary with response :rtype: str | dict **Example:** .. code-block:: python client.shiny.delete(shiny_id) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",can_be_none=False)Shiny._validate_type(shiny_id,"shiny_id",str,True)ifself._if_deployment_exist_for_asset(shiny_id):raiseWMLClientError("Cannot delete shiny asset that has existing deployments. Please delete all associated deployments and try again")response=requests.delete(self._client._href_definitions.get_asset_href(shiny_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,shiny_id:str|None=None,**kwargs:Any)->dict:"""Create a revision for the given shiny asset. Revisions are immutable once created. The metadata and attachment at `script_id` is taken and a revision is created out of it. :param shiny_id: ID of the shiny asset :type shiny_id: str :return: revised metadata of the stored shiny asset :rtype: dict **Example:** .. code-block:: python shiny_revision = client.shiny.create_revision(shiny_id) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",can_be_none=False)# For CP4D, check if either space or project ID is setself._client._check_if_either_is_set()Shiny._validate_type(shiny_id,"shiny_id",str,True)print("Creating shiny revision...")response=self._get_required_element_from_response(self._create_revision_artifact_for_assets(shiny_id,"Shiny"))final_response={"metadata":response["metadata"],"entity":{}}returnfinal_response

[docs]deflist_revisions(self,shiny_id:str|None=None,limit:int|None=None,**kwargs:Any)->DataFrame:"""List all revisions for the given shiny asset ID in a table format. :param shiny_id: ID of the stored shiny asset :type shiny_id: str :param limit: limit number of fetched records :type limit: int, optional :return: pandas.DataFrame with listed shiny revisions :rtype: pandas.DataFrame **Example:** .. code-block:: python client.shiny.list_revisions(shiny_id) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",can_be_none=False)# For CP4D, check if either space or project ID is setself._client._check_if_either_is_set()Shiny._validate_type(shiny_id,"shiny_id",str,True)url=self._client._href_definitions.get_asset_href(shiny_id)+"/revisions"# /v2/assets/{asset_id}/revisions returns 'results' objectshiny_resources=self._get_with_or_without_limit(url,None,"List Shiny revisions",summary=None,pre_defined=None,_all=self._should_get_all_values(limit),)["resources"]shiny_values=[(m["metadata"]["asset_id"],m["metadata"]["revision_id"],m["metadata"]["name"],m["metadata"]["commit_info"]["committed_at"],)forminshiny_resources]table=self._list(shiny_values,["ID","REVISION_ID","NAME","REVISION_COMMIT"],limit,)returntable

[docs]defget_revision_details(self,shiny_id:str|None=None,rev_id:str|None=None,**kwargs:Any)->ListType:"""Get metadata of the `shiny_id` revision. :param shiny_id: ID of the shiny asset :type shiny_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: stored shiny(s) metadata :rtype: list **Example:** .. code-block:: python shiny_details = client.shiny.get_revision_details(shiny_id, rev_id) """shiny_id=_get_id_from_deprecated_uid(kwargs,shiny_id,"shiny",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()Shiny._validate_type(shiny_id,"shiny_id",str,True)Shiny._validate_type(rev_id,"rev_id",str,False)ifrev_idisNone:rev_id="latest"url=self._client._href_definitions.get_asset_href(shiny_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=[{"metadata":response["metadata"],"entity":{}}forresponseinresponses]returnfinal_responses

def_get_required_element_from_response(self,response_data:dict)->dict:WMLResource._validate_type(response_data,"shiny",dict)revision_id=Nonemetadata={"guid":response_data["metadata"]["asset_id"],"href":response_data["href"],"name":response_data["metadata"]["name"],"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"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"]}if"description"inresponse_data["metadata"]:new_el["metadata"].update({"description":response_data["metadata"]["description"]})if"href"inresponse_data["metadata"]:href_without_host=response_data["metadata"]["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}")

Movatterモバイル変換

Source code for ibm_watsonx_ai.shiny