ModelArts Notebook动态挂载OBS和OBSFS桶脚本

import requests
import json
import os


class RomaMountOperation(object):
    authoring_url = "https://modelarts.cn-north-4.myhuaweicloud.com/v1/{}/notebooks".format(os.environ["PROJECT_ID"])

    def __init__(self, notebook_id):
        self.headers = {"Content-Type": "application/json",
                        "X-Auth-Token": self.get_roma_token_by_aksk()}

        self.mount_url = "{}/{}/storage".format(self.authoring_url, notebook_id)

    def get_roma_token_by_aksk(self):
        from modelarts.session import Session
        session =Session()
        body = {
            "auth": {
                "identity": {
                    "methods": [
                        "hw_ak_sk"
                    ],
                    "hw_ak_sk": {
                        "access": {
                            "key": session.access_key
                        },
                        "secret": {
                            "key": session.secret_key
                        }
                    }
                },
                "scope": {
                    "project": {
                        "id": session.project_id
                    }
                }
            }
        }
        headers = {
            "Content-Type": "application/json"
        }
        url = "https://iam.cn-north-4.myhuaweicloud.com/v3/auth/tokens"
        response = requests.post(url, headers=headers, data=json.dumps(body), verify=False)
        token = (response.headers['X-Subject-Token'])
        return token

    def mount(self, obs_uri):
        body = {
            "category": "OBS",
            "uri": obs_uri,  
            "mount_path": "/data/tmp/"
        }
        resp = requests.post(self.mount_url, headers=self.headers, data=json.dumps(body),
                             verify=False)
        print(json.dumps(resp.json(), indent=1))

    def unmount(self, storage_id):
        url = self.mount_url + "/" + storage_id
        print(url)
        response = requests.delete(url, headers=self.headers, verify=False)
        return response.json()
    
    def get(self, storage_id):
        url = self.mount_url + "/" + storage_id
        print(url)
        response = requests.get(url, headers=self.headers, verify=False)
        return response.json()
    
    def query(self):
        url = self.mount_url
        print(url)
        response = requests.get(url, headers=self.headers, verify=False)
        return response.json()


mount_instance = RomaMountOperation(notebook_id=os.environ["INSTANCE_ID"])   # notebook_id, 浏览器uri中的id
mount_instance.mount(obs_uri="obs://bucket/")            # 挂载：输入OBS地址
mount_instance.query()                                   # 查询挂载列表详情  
# mount_instance.get(storage_id="please input your storage id")
# mount_instance.unmount(storage_id="please input your storage id")