Docs

Database Transmission Service

Python SDK

Code examples

This topic describes how to use the Database Transmission Service (DTS) Python SDK to quickly call APIs to create data transmission tasks and query task information.

Prerequisites

Create data transmission task

MySQL

Redis

PostgreSQL

MongoDB

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", 
                    "AccountMapping":{ 
					              "Account": "test****", 
					              "ResetPassword": "true", 
					              "Password": "*Test****" 
					            } 
                    } 
                }, 
                "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' 
    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': task_name, 
        'TaskType': task_type, 
        'ChargeConfig': { 
            'ChargeType': 'PostPaid', 
            'OneStep': True 
        }, 
        'SrcConfig': { 
            'EndpointType': 'Public_Redis', 
            'PublicRedisSettings': { 
                'Host': 'IP_str_here', 
                'Port': 6379, 
                'Username': 'default', 
                'Password': 'redis_password_here', 
                'RegionSettings': { 
                    'Region': region_str 
                } 
            } 
        }, 
        'DestConfig': { 
            'EndpointType': 'Volc_Redis', 
            'VolcRedisSettings': { 
                'DBInstanceId': 'redis-instanceID', 
                'Username': 'default', 
                'Password': 'redis_password_here', 
                'RegionSettings': { 
                    'Region': region_str 
                } 
            } 
        }, 
        'SolutionSettings': { 
            'SolutionType': 'Redis2Redis', 
            'Redis2RedisSettings': { 
                'ObjectMappings': [ 
                    { 
                        "DestObjName": "0", 
                        "ObjectType": "Database", 
                        "SrcObjName": "0" 
                    }, 
                    { 
                        "DestObjName": "6", 
                        "ObjectType": "Database", 
                        "SrcObjName": "6" 
                    } 
                ], 
                'FullTransmissionSettings': { 
                    'EnableFull': True 
                }, 
                'IncrTransmissionSettings': { 
                    'EnableIncr': True 
                }, 
                'ErrorBehaviorSettings': { 
                    'MaxRetrySeconds': 7200 
                } 
            } 
        } 
    } 
    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' 
    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": "rudonxxxxx-1", 
		    "TaskType": "DataMigration", 
		    "TrafficSpec": "Compact", 
		    "ProjectName": "default", 
		    "Tags": [ 
               { 
                   "Key": "key1", 
                   "Value": "value1" 
               } 
           ], 
		     "SrcConfig": { 
		         "EndpointType": "Public_PostgreSQL", 
		         "PublicPostgreSQLSettings": { 
		             "Host": "192.168.***.***", 
		             "Port": 4, 
		             "Username": "test****", 
		             "Password": "Admin@****", 
		             "DBName": "test****", 
		             "RegionSettings": { 
		                "Region": "cn-beijing" 
		            } 
		         } 
			     }, 
		     "DestConfig": { 
		         "EndpointType": "Volc_PostgreSQL", 
		         "VolcPostgreSQLSettings": { 
		             "DBInstanceId": "postgres-h441603c68aaa****", 
		             "Username": "test****", 
		             "Password": "Admin@****", 
		             "DBName": "test****", 
		             "RegionSettings": { 
		                "Region": "cn-beijing" 
		            } 
		         }, 
		     }, 
		    "SolutionSettings": { 
		        "SolutionType": "PG2PG", 
		        "AutoStart": True, 
		        "PG2PGSettings": { 
		             "ObjectMappings": [ 
		                 { 
		                    "ObjectType": "Database", 
		                    "SrcObjName": "0", 
		                    "DestObjName": "0" 
		                  }, 
		                { 
		                    "ObjectType": "Table", 
		                    "SrcObjName": "1", 
		                    "DestObjName": "1" 
		                  }, 
		                { 
		                    "ObjectType": "View", 
		                    "SrcObjName": "2", 
		                    "DestObjName": "2" 
		                 }, 
		                { 
		                    "ObjectType": "PostSequence", 
		                    "SrcObjName": "3", 
		                    "DestObjName": "3" 
		                } 
		            ], 
		             "MetaTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		             "FullTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		             "IncrTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		              "ErrorBehaviorSettings": { 
		                "MaxRetrySeconds": 7200 
		            } 
				    } , 
			      "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' 
    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": "rudonxxxxx-1", 
		    "TaskType": "DataMigration", 
		    "TrafficSpec": "Compact", 
		    "ProjectName": "default", 
		    "Tags": [ 
		               { 
		                   "Key": "key1", 
		                   "Value": "value1" 
		               } 
		           ], 
		     "SrcConfig": { 
		         "EndpointType": "Public_Mongo", 
		         "PublicMongoSettings": { 
		             "Endpoints": "192.168.***.***:4", 
		             "Username": "test****", 
		             "Password": "Admin@****", 
		             "MongoAuthSourceDB": "ReplicaSet", 
		             "DeployType": "Standalone", 
		             "RegionSettings": { 
		                "Region": "cn-beijing" 
		            } 
		         } 
		     }, 
		     "DestConfig": { 
		         "EndpointType": "Volc_Mongo", 
		         "VolcMongoSettings": { 
		             "DBInstanceId": "mongo-replica-e405f8e2****", 
		             "Username": "test****", 
		             "Password": "Admin@****", 
		             "MongoAuthSourceDB": "admin", 
		             "DeployType": "ShardedCluster", 
		             "RegionSettings": { 
		                "Region": "cn-beijing" 
		            } 
		         }, 
		     
		     }, 
		    "SolutionSettings": { 
		        "SolutionType": "Mongo2Mongo", 
		        "AutoStart": True, 
		        "Mongo2MongoSettings": { 
		             "ObjectMappings": [ 
		                { 
		                    "ObjectType": "Table", 
		                    "SrcObjName": "0", 
		                    "DestObjName": "0" 
		                }, 
		                { 
		                    "ObjectType": "Aggregate", 
		                    "SrcObjName": "1", 
		                    "DestObjName": "1" 
		                } 
		            ], 
		             "MetaTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		             "FullTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		             "IncrTransmissionSettings": { 
		                 "EnableMeta": True 
		             }, 
		              "ErrorBehaviorSettings": { 
		                "MaxRetrySeconds": 7200 
		            } 
		    } , 
		      "ChargeConfig":{ 
		        "ChargeType":"PostPaid" 
		    } 
    } 
    resp = dts_service.create_transmission_task(params, body) 
    print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))

Query data transmission task information

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=(',', ':')))

Pause data transmission task

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=(',', ':')))

Start data transmission task

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=(',', ':')))

Retry data transmission task

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=(',', ':')))

Recover data transmission task

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=(',', ':')))

Terminate data transmission task

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=(',', ':')))

Delete data transmission task

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=(',', ':')))

Create pre-check task

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.pre_check_async(params, body) 
    print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))

Query pre-check results

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' 
    id_str = 'id_str'  # we get id_str from precheck_async api 

    dts_service = DtsService(region=region_str) 

    dts_service.set_ak(access_key) 
    dts_service.set_sk(secret_key) 

    params = {} 
    body = { 
        'ID': id_str 
    } 

    resp = dts_service.get_async_pre_check_result(params, body) 
    print(json.dumps(resp, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ':')))

FAQs

If there is a Can't connect to HTTPS URL when sending a request using the Python SDK because the SSL module is not available error, What should I do?

Possible cause: the SSL module was not installed correctly when the Python SDK was installed.

Solution: reinstall Python and ensure that the SSL module is included during the installation process.

Last updated: 2026.01.21 17:37:45

Database Transmission Service

Prerequisites #

Create data transmission task #

Query data transmission task information #

Pause data transmission task #

Start data transmission task #

Retry data transmission task #

Recover data transmission task #

Terminate data transmission task #

Delete data transmission task #

Create pre-check task #

Query pre-check results #

FAQs #

If there is a Can't connect to HTTPS URL when sending a request using the Python SDK because the SSL module is not available error, What should I do? #