Movatterモバイル変換

[docs]classProjects(WMLResource):"""Store and manage projects. .. note:: Projects module is available since Python SDK version 1.3.5. """ConfigurationMetaNames=ProjectsMetaNames()"""MetaNames for projects creation."""MemberMetaNames=MemberMetaNames()"""MetaNames for project members creation."""def__init__(self,client:APIClient):WMLResource.__init__(self,__name__,client)self._client=clientdef_get_resources(self,url:str,op_name:str,params:dict|None=None)->dict:ifparamsisnotNoneand"limit"inparams.keys():ifparams["limit"]<1:raiseWMLClientError("Limit cannot be lower than 1.")elifparams["limit"]>1000:raiseWMLClientError("Limit cannot be larger than 1000.")ifparamsisnotNoneandlen(params)>0:response_get=requests.get(url,headers=self._client._get_headers(),params=params)returnself._handle_response(200,op_name,response_get)else:resources=[]whileTrue:response_get=requests.get(url,headers=self._client._get_headers())result=self._handle_response(200,op_name,response_get)resources.extend(result["resources"])if"next"notinresult:breakelse:url=self._credentials.url+result["next"]["href"]if"start=invalid"inurl:breakreturn{"resources":resources}

[docs]defstore(self,meta_props:dict)->dict:"""Create a project. :param meta_props: metadata of the project configuration. To see available meta names, use: .. code-block:: python client.projects.ConfigurationMetaNames.get() :type meta_props: dict :return: metadata of the stored project :rtype: dict **Example:** .. code-block:: python meta_props = { client.projects.ConfigurationMetaNames.NAME: "my project", client.projects.ConfigurationMetaNames.DESCRIPTION: "test project", client.projects.ConfigurationMetaNames.STORAGE: { "type": "assetfiles" } } projects_details = client.projects.store(meta_props) """Projects._validate_type(meta_props,"meta_props",dict,True)ifself.ConfigurationMetaNames.GENERATORnotinmeta_props:meta_props[self.ConfigurationMetaNames.GENERATOR]="Watsonx-Python-SDK"if"compute"inmeta_props:if"name"notinmeta_props["compute"]:raiseWMLClientError("'name' is mandatory for 'COMPUTE'")if"type"notinmeta_props["compute"]:meta_props["compute"]["type"]="machine_learning"project_meta=self.ConfigurationMetaNames._generate_resource_metadata(meta_props,with_validation=True,client=self._client)if"compute"inproject_meta:project_meta["compute"]=[project_meta["compute"]]creation_response=requests.post(self._client._href_definitions.get_transactional_projects_href(),headers=self._client._get_headers(),json=project_meta,)location=self._handle_response(201,"creating new project",creation_response,_silent_response_logging=True,)["location"]project_details=self.get_details(location.split("/")[-1])if"compute"inproject_details["entity"].keys():instance_id=project_details["entity"]["compute"][0]["guid"]self._client.service_instance=ServiceInstance(self._client)self._client.service_instance._instance_id=instance_idreturnproject_details

[docs]@staticmethoddefget_id(project_details:dict)->str:"""Get the project_id from the project details. :param project_details: metadata of the stored project :type project_details: dict :return: ID of the stored project :rtype: str **Example:** .. code-block:: python project_details = client.projects.store(meta_props) project_id = client.projects.get_id(project_details) """Projects._validate_type(project_details,"project_details",object,True)returnWMLResource._get_required_element_from_dict(project_details,"project_details",["metadata","guid"])

[docs]defget_id_by_name(self,project_name:str)->str:"""Get the ID of a stored project by name. :param project_name: name of the stored project :type project_name: str :return: ID of the stored project :rtype: str **Example:** .. code-block:: python project_id = client.projects.get_id_by_name(project_name) """Projects._validate_type(project_name,"project_name",str,True)details=self.get_details(project_name=project_name)["resources"]iflen(details)>1:raiseMultipleResourceIdByNameFound(project_name,"project")eliflen(details)==0:raiseResourceIdByNameNotFound(project_name,"project")returnself.get_id(details[0])

[docs]defdelete(self,project_id:str)->Literal["SUCCESS"]:"""Delete a stored project. :param project_id: ID of the project :type project_id: str :return: status "SUCCESS" if deletion is successful :rtype: Literal["SUCCESS"] **Example:** .. code-block:: python client.projects.delete(project_id) """Projects._validate_type(project_id,"project_id",str,True)project_endpoint=(self._client._href_definitions.get_transactional_project_href(project_id))response_delete=requests.delete(project_endpoint,headers=self._client._get_headers())response=self._handle_response(204,"project deletion",response_delete,False)print("DELETED")return"SUCCESS"

[docs]defget_details(self,project_id:str|None=None,limit:int|None=None,asynchronous:bool|None=False,get_all:bool|None=False,project_name:str|None=None,**kwargs:Any,)->dict:"""Get metadata of stored project(s). :param project_id: ID of the project :type project_id: str, optional :param limit: applicable when `project_id` is not provided, otherwise `limit` will be ignored :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 :param project_name: name of the stored project, can be used only when `project_id` is None :type project_name: str, optional :return: metadata of stored project(s) :rtype: - **dict** - if project_id is not None - **{"resources": [dict]}** - if project_id is None **Example:** .. code-block:: python project_details = client.projects.get_details(project_id) project_details = client.projects.get_details(project_name) project_details = client.projects.get_details(limit=100) project_details = client.projects.get_details(limit=100, get_all=True) project_details = [] for entry in client.projects.get_details(limit=100, asynchronous=True, get_all=True): project_details.extend(entry) """Projects._validate_type(project_id,"project_id",str,False)href=self._client._href_definitions.get_project_href(project_id)query_params={}ifinclude:=kwargs.get("include"):query_params["include"]=includeifproject_idisnotNone:response_get=requests.get(href,headers=self._client._get_headers(),params=query_params)returnself._handle_response(200,"Get project",response_get,_silent_response_logging=True)ifproject_name:query_params.update({"name":project_name})returnself._get_with_or_without_limit(self._client._href_definitions.get_projects_href(),100ifnotlimitorlimit>100elselimit,"projects",summary=False,pre_defined=False,skip_space_project_chk=True,query_params=query_params,_async=asynchronous,_all=get_all,_silent_response_logging=True,)

@cached(cache=TTLCache(maxsize=32,ttl=4.5*60))# Projects API doesn't refresh credentials until 5 minutes before expirationdef_get_details(self,project_id:str|None=None,limit:int|None=None,asynchronous:bool|None=False,get_all:bool|None=False,project_name:str|None=None,**kwargs:Any,)->dict:"""Get metadata of stored project(s) with caching. It's dedicated for internal usage."""returnself.get_details(project_id=project_id,limit=limit,asynchronous=asynchronous,get_all=get_all,project_name=project_name,**kwargs,)

[docs]deflist(self,limit:int|None=None,member:str|None=None,roles:str|None=None,project_type:str|None=None,)->DataFrame:"""List stored projects in a table format. :param limit: limit number of fetched records :type limit: int, optional :param member: filters the result list, only includes projects where the user with a matching user ID is a member :type member: str, optional :param roles: a list of comma-separated project roles to use to filter the query results, must be used in conjunction with the "member" query parameter, available values : `admin`, `editor`, `viewer` :type roles: str, optional :param project_type: filter projects by their type, available types are 'cpd', 'wx', 'wca', 'dpx' and 'wxbi' :type project_type: str, optional :return: pandas.DataFrame with listed projects :rtype: pandas.DataFrame **Example:** .. code-block:: python client.projects.list() """Projects._validate_type(limit,"limit",int,False)href=self._client._href_definitions.get_projects_href()params:dict[str,Any]={}limit=100ifnotlimitorlimit>100elselimitifmemberisnotNone:params.update({"member":member})ifrolesisnotNone:params.update({"roles":roles})ifproject_typeisnotNone:params.update({"type":project_type})projects_resources=[mforrinself._get_with_or_without_limit(href,limit,"projects",summary=False,pre_defined=False,skip_space_project_chk=True,query_params=params,_async=True,_all=True,_silent_response_logging=True,)forminr["resources"]]project_values=[(m["metadata"]["guid"],m["entity"]["name"],m["metadata"]["created_at"])forminprojects_resources]table=self._list(project_values,["ID","NAME","CREATED"],limit)returntable

[docs]defupdate(self,project_id:str,changes:dict)->dict:"""Update existing project metadata. 'STORAGE' cannot be updated. :param project_id: ID of the project with the definition to be updated :type project_id: str :param changes: elements to be changed, where keys are ConfigurationMetaNames :type changes: dict :return: metadata of the updated project :rtype: dict **Example:** .. code-block:: python metadata = { client.projects.ConfigurationMetaNames.NAME:"updated_project", client.projects.ConfigurationMetaNames.COMPUTE: {"name": "test_instance", "crn": "v1:staging:public:pm-20-dev:us-south:a/09796a1b4cddfcc9f7fe17824a68a0f8:f1026e4b-77cf-4703-843d-c9984eac7272::" } } project_details = client.projects.update(project_id, changes=metadata) """if"storage"inchanges:raiseWMLClientError("STORAGE cannot be updated")if"generator"inchanges:raiseWMLClientError("GENERATOR cannot be updated")if"scope"inchanges:raiseWMLClientError("SCOPE cannot be updated")if"creator"inchanges:raiseWMLClientError("creator cannot be updated")if"creator_iam_id"inchanges:raiseWMLClientError("creator_iam_id cannot be updated")self._validate_type(project_id,"project_id",str,True)self._validate_type(changes,"changes",dict,True)details=self.get_details(project_id)if"compute"inchanges:changes["compute"]["type"]="machine_learning"payload_compute=[]payload_compute.append(changes["compute"])changes["compute"]=payload_computepatch_payload=self.ConfigurationMetaNames._generate_patch_payload(details["entity"],changes)payload=details["entity"]defmodify(tree,path,value):iflen(path)>1:modify(tree[path[0]],path[1:],value)else:tree[path[0]]=valueforrinpatch_payload:path=r["path"].strip("/").split("/")modify(payload,path,r["value"])forkeyin["storage","generator","scope","creator","creator_iam_id"]:ifkeyinpayload:payload.pop(key)href=self._client._href_definitions.get_project_href(project_id)response=requests.patch(href,json=payload,headers=self._client._get_headers())updated_details=self._handle_response(200,"projects patch",response,_silent_response_logging=True)# Cloud Convergenceif"compute"inupdated_details["entity"].keys():instance_id=updated_details["entity"]["compute"][0]["guid"]self._client.service_instance=ServiceInstance(self._client)self._client.service_instance._instance_id=instance_idreturnupdated_details

#######SUPPORT FOR PROJECT MEMBERS

[docs]defcreate_member(self,project_id:str,meta_props:dict)->dict:"""Create a member within a project. :param project_id: ID of the project with the definition to be updated :type project_id: str :param meta_props: metadata of the member configuration. To see available meta names, use: .. code-block:: python client.projects.MemberMetaNames.get() :type meta_props: dict :return: metadata of the stored member :rtype: dict .. note:: * `role` can be any one of the following: "viewer", "editor", "admin" * `type` can be any one of the following: "user", "service" * `id` can be one of the following: service-ID or IAM-userID **Examples** .. code-block:: python metadata = { client.projects.MemberMetaNames.MEMBERS: [{"id":"IBMid-100000DK0B", "type": "user", "role": "admin" }] } members_details = client.projects.create_member(project_id=project_id, meta_props=metadata) .. code-block:: python metadata = { client.projects.MemberMetaNames.MEMBERS: [{"id":"iam-ServiceId-5a216e59-6592-43b9-8669-625d341aca71", "type": "service", "role": "admin" }] } members_details = client.projects.create_member(project_id=project_id, meta_props=metadata) """self._validate_type(project_id,"project_id",str,True)Projects._validate_type(meta_props,"meta_props",dict,True)meta={}if"members"inmeta_props:meta=meta_propselif"member"inmeta_props:dictionary=meta_props["member"]payload=[]payload.append(dictionary)meta["members"]=payloadproject_meta=self.MemberMetaNames._generate_resource_metadata(meta,with_validation=True,client=self._client)creation_response=requests.post(self._client._href_definitions.get_projects_members_href(project_id),headers=self._client._get_headers(),json=project_meta,)members_details=self._handle_response(200,"creating new members",creation_response)returnmembers_details

[docs]defget_member_details(self,project_id:str,user_name:str|None=None)->dict:"""Get metadata of a member associated with a project. If no user_name is passed, all members details will be returned. :param project_id: ID of that project with the definition to be updated :type project_id: str :param user_name: name of the member :type user_name: str, optional :return: metadata of the project member :rtype: dict **Example:** .. code-block:: python member_details = client.projects.get_member_details(project_id, "test@ibm.com") members_details = client.projects.get_member_details(project_id) """Projects._validate_type(project_id,"project_id",str,True)Projects._validate_type(user_name,"member_id",str,False)ifuser_name:href=self._client._href_definitions.get_projects_member_href(project_id,user_name)response_get=requests.get(href,headers=self._client._get_headers())returnself._handle_response(200,"Get project member",response_get)else:href=self._client._href_definitions.get_projects_members_href(project_id)response_get=requests.get(href,headers=self._client._get_headers())returnself._handle_response(200,"Get project members",response_get)

[docs]defdelete_member(self,project_id:str,user_name:str|None=None)->str:"""Delete a member associated with a project. :param project_id: ID of the project :type project_id: str :param user_name: name of the member :type user_name: str, optional :return: status ("SUCCESS" if succeeded) :rtype: str **Example:** .. code-block:: python client.projects.delete_member(project_id, user_name) """Projects._validate_type(project_id,"project_id",str,True)Projects._validate_type(user_name,"user_name",str,False)member_endpoint=self._client._href_definitions.get_projects_member_href(project_id,user_name)response_delete=requests.delete(member_endpoint,headers=self._client._get_headers())print("DELETED")self._handle_response(204,"project member deletion",response_delete,False)return"SUCCESS"

[docs]defupdate_member(self,project_id:str,user_name:str,changes:dict)->dict:"""Update the metadata of an existing member. :param project_id: ID of the project :type project_id: str :param user_name: name of the member to be updated :type user_name: str :param changes: elements to be changed, where keys are ConfigurationMetaNames :type changes: dict :return: metadata of the updated member :rtype: dict **Example:** .. code-block:: python metadata = { client.projects.MemberMetaNames.MEMBER: {"role": "editor"} } member_details = client.projects.update_member(project_id, user_name, changes=metadata) """self._validate_type(project_id,"project_id",str,True)self._validate_type(user_name,"user_name",str,True)self._validate_type(changes,"changes",dict,True)user_details=self.get_member_details(project_id,user_name)patch_request=[]deluser_details["type"]deluser_details["state"]user_details.update(changes["member"])patch_request.append(user_details)# patching is different here, you just pass updated members but without `state` and `type`response=requests.patch(self._client._href_definitions.get_projects_members_href(project_id),json={"members":patch_request},headers=self._client._get_headers(),)updated_details=self._handle_response(200,"members patch",response)returnupdated_details

[docs]deflist_members(self,project_id:str,limit:int|None=None,identity_type:str|None=None,role:str|None=None,state:str|None=None,)->DataFrame:"""Print the stored members of a project in a table format. :param project_id: ID of the project :type project_id: str :param limit: limit number of fetched records :type limit: int, optional :param identity_type: filter the members by type :type identity_type: str, optional :param role: filter the members by role :type role: str, optional :param state: filter the members by state :type state: str, optional :return: pandas.DataFrame with listed members :rtype: pandas.DataFrame **Example:** .. code-block:: python client.projects.list_members(project_id) """self._validate_type(project_id,"project_id",str,True)params:dict[str,Any]={}iflimitisnotNone:params.update({"limit":limit})ifidentity_typeisnotNone:params.update({"type":identity_type})ifroleisnotNone:params.update({"role":role})ifstateisnotNone:params.update({"state":state})href=self._client._href_definitions.get_projects_members_href(project_id)member_resources=self._get_resources(href,"project members",params)["resources"]project_values=[((m["id"],m["type"],m["role"],m["state"])if"state"inmelse(m["id"],m["type"],m["role"],None))forminmember_resources]table=self._list(project_values,["ID","TYPE","ROLE","STATE"],limit)returntable

Movatterモバイル変換

Source code for ibm_watsonx_ai.projects