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

举报
modelarts-dev-server 发表于 2024/03/21 16:58:41 2024/03/21
【摘要】 在ModelArts Notbeook动态挂载OBS桶和OBSFS桶的API脚本

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")
【版权声明】本文为华为云社区用户原创内容,未经允许不得转载,如需转载请自行联系原作者进行授权。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。