最近更新时间：2023.07.25 10:23:59
首次发布时间：2023.07.18 16:13:48
本文介绍如何使用数据库传输服务 DTS Python SDK 快速调用 API 执行创建数据传输任务和查询任务信息。
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' task_name = 'task_name_str' task_type = 'task_type_here' region_str = 'region_str_here' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { "TaskName":"zaizaitest****1", "TaskType":"DataMigration", "SrcConfig": { "EndpointType": "Public_MySQL", "PublicMySQLSettings": { "Host": "192.168.***.***", "Port": 4, "Username": "test****", "Password": "Admin@****", "SSLSettings": { "EnableSSL": "true", "Cert": "-----BEGIN CERTIFICATE----- ...... -----END CERTIFICATE-----" }, "RegionSettings": { "Region": "cn-shanghai" } } }, "DestConfig": { "EndpointType": "Volc_MySQL", "VolcMySQLSettings": { "DBInstanceId": "mysql-7cd1dc4ab511", "DBInstanceName": "", "Password": "Admin@****", "RegionSettings": { "Region": "cn-guangzhou" }, "Username": "test****" } }, "TrafficSpec":"Compact", "ProjectName":"default", "SolutionSettings": { "MySQL2MySQLSettings": { "AccountTransmissionSettings": { "EnableAccount": true }, "ETLSettings": { "Script": "" }, "ErrorBehaviorSettings": { "MaxRetrySeconds": 7200 }, "FullTransmissionSettings": { "EnableFull": true, "Snapshot": false }, "IncrTransmissionSettings": { "EnableIncr": true, "Statements": [ "StmtDDLAll", "StmtDMLDelete", "StmtDMLUpdate", "StmtDMLInsert" ] }, "MetaTransmissionSettings": { "EnableMeta": true }, "ObjectMappings": [ { "DestObjName": "huahua", "MappingList": [ { "DestObjName": "*", "ObjectType": "Table", "SrcObjName": "*" } ], "ObjectType": "Database", "SrcObjName": "huahua" } ], "PolicyForPrimaryKeyConflict": "Override" }, "SolutionType": "MySQL2MySQL" }, "Tags": [ { "Key": "key1", "Value": "value1" } ], "ChargeConfig":{ "ChargeType":"PostPaid" } } resp = dts_service.create_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.describe_transmission_task_info(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.suspend_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.start_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.retry_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.resume_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.stop_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))
import json from volcengine.dts.dts_service import DtsService if __name__ == '__main__': access_key = 'your_ak_here' secret_key = 'your_sk_here' region_str = 'region_str_here' task_id_str = 'task_id_str' dts_service = DtsService(region=region_str) dts_service.set_ak(access_key) dts_service.set_sk(secret_key) params = {} body = { 'TaskId': task_id_str } resp = dts_service.delete_transmission_task(params, body) print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))