存储迁移服务
存储迁移服务
存储迁移服务
文档指南
请输入
产品计费
用户指南
API 参考
DMS 1.0 API
DMS 2.0 API
openapi
SDK 参考
DMS 1.0 SDK
错误码
1002-执行错误
1004-目标端错误
1005-专线错误
相关协议
- 文档首页
存储迁移服务SDK 参考DMS 1.0 SDKPython SDK
复制全文
下载 pdfPython SDK
文档反馈
问问助手本文介绍 Python SDK 的注意事项、安装方式和初始化示例。
注意事项
- Python 需要至少 2.7 或以上版本。
- 由于 Windows 系统有最长路径限制，可能会导致安装失败，请按照以下方式设置：
- 按下 Win+R ，输入 regedit 打开注册表编辑器。
- 设置 \HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\FileSystem 路径下的变量 LongPathsEnabled 为 1。
SDK 地址
安装 Python SDK
执行以下命令，通过 pip 安装 Python SDK。
pip install volcengine-python-sdk
配置长期访问凭证
初始化 SDK 前，您需要先配置长期访问凭证。本章节介绍不同操作系统下配置长期访问凭证的操作步骤。
macOS
Linux
Windows
- 打开终端并执行以下命令打开文件。
nano ~/.bash_profile
- 在文件末尾添加 AKSK 信息。
export TOS_ACCESS_KEY=AKTPYmI1Z****
export TOS_SECRET_KEY=T1dJM01UU****
- 保存文件并退出。
- 执行以下命令生效配置信息。
source ~/.bash_profile
- 执行以下命令验证配置信息。
echo $TOS_ACCESS_KEY
echo $TOS_SECRET_KEY
如果配置成功，则返回如下示例：
AKTPYmI1Z****
T1dJM01UU****
- 打开终端并执行以下命令打开文件。
sudo vim /etc/profile
- 在文件末尾添加 AKSK 信息。
export TOS_ACCESS_KEY=AKTPYmI1Z****
export TOS_SECRET_KEY=T1dJM01UU****
- 保存文件并退出。
- 执行以下命令生效配置信息。
source /etc/profile
- 执行以下命令验证配置信息。
echo $TOS_ACCESS_KEY
echo $TOS_SECRET_KEY
如果配置成功，则返回如下示例：
AKTPYmI1Z****
T1dJM01UU****
- 打开命令行工具，执行以下命令，添加 AKSK 信息。
set TOS_ACCESS_KEY=AKTPYmI1Z****
set TOS_SECRET_KEY=T1dJM01UU****
- 执行以下命令生效配置信息。
setx TOS_ACCESS_KEY "%TOS_ACCESS_KEY%"
setx TOS_SECRET_KEY "%TOS_SECRET_KEY%"
- 重启命令行工具，然后执行以下命令验证配置信息。
echo %TOS_ACCESS_KEY%
echo %TOS_SECRET_KEY%
如果配置成功，则返回如下示例：
AKTPYmI1Z****
T1dJM01UU****
初始化示例
from __future__ import print_function
import os
import volcenginesdkdms
import volcenginesdkcore
from pprint import pprint
from volcenginesdkcore.rest import ApiException
if __name__ == '__main__':
configuration = volcenginesdkcore.Configuration()
configuration.ak = os.getenv("AK")
configuration.sk = os.getenv("SK")
configuration.region = "the region your target bucket in"
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)
# use global default configuration
api_instance = volcenginesdkdms.DMSApi()
# use custom configuration
# api_instance = volcenginesdkecs.ECSApi(volcenginesdkcore.ApiClient(configuration))
try:
resp = api_instance.list_data_migrate_task(volcenginesdkdms.ListDataMigrateTaskRequest(offset=0, limit=10))
pprint(resp)
except ApiException as e:
print("Exception when calling DMSApi->list_data_migrate_task: %s\n" % e)
try:
resp = api_instance.create_data_migrate_task(
volcenginesdkdms.CreateDataMigrateTaskRequest(
basic_config=volcenginesdkdms.BasicConfigForCreateDataMigrateTaskInput(
task_name="task-name",
overwrite_policy="None", # "Force", "None", "LastModify"
source_type="StorageTypeObject", # "StorageTypeObject", "StorageTypeUrl"
storage_class="Standard", # "Standard", "InheritSource", "Ia", "ArchiveFr",
# "IntelligentTiering", "ColdArchive", "Archive", "DeepColdArchive"
failed_num_to_abort=-1,
),
source=volcenginesdkdms.SourceForCreateDataMigrateTaskInput(
object_source_config=volcenginesdkdms.ObjectSourceConfigForCreateDataMigrateTaskInput(
bucket_access_config=volcenginesdkdms.BucketAccessConfigForCreateDataMigrateTaskInput(
ak="source bucket ak",
sk="source bucket sk",
bucket_name="",
region=configuration.region,
vendor="StorageVendorS3",
endpoint="",
),
)
),
target=volcenginesdkdms.TargetForCreateDataMigrateTaskInput(
bucket_name="",
ak="target bucket ak",
sk="target bucket sk",
)
))
pprint(resp)
except ApiException as e:
print("Exception when calling DMSApi->create_data_migrate_task: %s\n" % e)
创建 URL 列表迁移任务
from __future__ import print_function
import os
import volcenginesdkdms
import volcenginesdkcore
from pprint import pprint
from volcenginesdkcore.rest import ApiException
if __name__ == '__main__':
configuration = volcenginesdkcore.Configuration()
configuration.ak = os.getenv("AK")
configuration.sk = os.getenv("SK")
configuration.region = "the region your target bucket in"
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)
# use global default configuration
api_instance = volcenginesdkdms.DMSApi()
# use custom configuration
# api_instance = volcenginesdkecs.ECSApi(volcenginesdkcore.ApiClient(configuration))
try:
resp = api_instance.list_data_migrate_task(volcenginesdkdms.ListDataMigrateTaskRequest(offset=0, limit=10))
pprint(resp)
except ApiException as e:
print("Exception when calling DMSApi->list_data_migrate_task: %s\n" % e)
try:
resp = api_instance.create_data_migrate_task(volcenginesdkdms.CreateDataMigrateTaskRequest(
basic_config=volcenginesdkdms.BasicConfigForCreateDataMigrateTaskInput(task_name="task-name",
overwrite_policy="None", # "Force", "None", "LastModify"
source_type="StorageTypeUrl", # "StorageTypeObject", "StorageTypeUrl"
storage_class="Standard", # "Standard", "InheritSource", "Ia", "ArchiveFr",
# "IntelligentTiering", "ColdArchive", "Archive", "DeepColdArchive"
failed_num_to_abort=-1,
), source=volcenginesdkdms.SourceForCreateDataMigrateTaskInput(
url_source_config=volcenginesdkdms.UrlSourceConfigForCreateDataMigrateTaskInput(
url_list_name="url_list.txt", # 存放在桶中的 URL 清单的对象名
bucket_access_config=volcenginesdkdms.BucketAccessConfigForCreateDataMigrateTaskInput(
ak="url bucket ak", sk="url bucket sk", bucket_name="url bucket name",
region=configuration.region, vendor="StorageVendorTOS", endpoint="tos endpoint"))),
target=volcenginesdkdms.TargetForCreateDataMigrateTaskInput(bucket_name="", ak="target bucket ak",
sk="target bucket sk", )))
pprint(resp)
except ApiException as e:
print("Exception when calling DMSApi->create_data_migrate_task: %s\n" % e)
最近更新时间：2026.01.27 18:52:32
这个页面对您有帮助吗？
有用
有用
无用
无用