feat: add workflow template id for pfop

Author: lihsai0

.gitignore

@@ -36,6 +36,7 @@ pip-log.txt
 .coverage
 .tox
 nosetests.xml
+coverage.xml
 
 # Translations
 *.mo
@@ -45,4 +46,4 @@ nosetests.xml
 .project
 .pydevproject
 /.idea
-/.venv
+/.venv*

qiniu/auth.py

@@ -34,10 +34,11 @@
     str('fsizeMin'),  # 上传文件最少字节数
     str('keylimit'),  # 设置允许上传的key列表，字符串数组类型，数组长度不可超过20个，如果设置了这个字段，上传时必须提供key
 
-    str('persistentOps'),  # 持久化处理操作
+    str('persistentOps'),  # 持久化处理操作，与 persistentWorkflowTemplateID 二选一
     str('persistentNotifyUrl'),  # 持久化处理结果通知URL
     str('persistentPipeline'),  # 持久化处理独享队列
     str('persistentType'),  # 为 `1` 时，开启闲时任务，必须是 int 类型
+    str('persistentWorkflowTemplateID'),  # 工作流模板 ID，与 persistentOps 二选一
 
     str('deleteAfterDays'),  # 文件多少天后自动删除
     str('fileType'),  # 文件的存储类型，0为标准存储，1为低频存储，2为归档存储，3为深度归档存储，4为归档直读存储

qiniu/services/processing/pfop.py

@@ -24,36 +24,47 @@ def __init__(self, auth, bucket, pipeline=None, notify_url=None):
         self.pipeline = pipeline
         self.notify_url = notify_url
 
-    def execute(self, key, fops, force=None, persistent_type=None):
+    def execute(self, key, fops=None, force=None, persistent_type=None, workflow_template_id=None):
         """
         执行持久化处理
 
         Parameters
         ----------
         key: str
             待处理的源文件
-        fops: list[str]
+        fops: list[str], optional
             处理详细操作，规格详见 https://developer.qiniu.com/dora/manual/1291/persistent-data-processing-pfop
+            与 template_id 二选一
         force: int or str, optional
             强制执行持久化处理开关
         persistent_type: int or str, optional
             持久化处理类型，为 '1' 时开启闲时任务
+        template_id: str, optional
+            与 fops 二选一
         Returns
         -------
         ret: dict
             持久化处理的 persistentId，类似 {"persistentId": 5476bedf7823de4068253bae};
         resp: ResponseInfo
         """
-        ops = ';'.join(fops)
-        data = {'bucket': self.bucket, 'key': key, 'fops': ops}
+        if not fops and not workflow_template_id:
+            raise ValueError('Must provide one of fops or template_id')
+        data = {
+            'bucket': self.bucket,
+            'key': key,
+        }
         if self.pipeline:
             data['pipeline'] = self.pipeline
         if self.notify_url:
             data['notifyURL'] = self.notify_url
+        if fops:
+            data['fops'] = ';'.join(fops)
         if force == 1 or force == '1':
             data['force'] = str(force)
         if persistent_type and type(int(persistent_type)) is int:
             data['type'] = str(persistent_type)
+        if workflow_template_id:
+            data['workflowTemplateID'] = workflow_template_id
 
         url = '{0}/pfop'.format(config.get_default('default_api_host'))
         return http._post_with_auth(url, data, self.auth)

tests/cases/test_services/test_processing/test_pfop.py

@@ -1,6 +1,5 @@
 import pytest
 
-
 from qiniu import PersistentFop, op_save
 
 
@@ -16,6 +15,7 @@ def test_pfop_execute(self, qn_auth):
         ]
         ret, resp = pfop.execute('sintel_trailer.mp4', ops, 1)
         assert resp.status_code == 200, resp
+        assert ret is not None, resp
         assert ret['persistentId'] is not None, resp
         global persistent_id
         persistent_id = ret['persistentId']
@@ -27,23 +27,71 @@ def test_pfop_get_status(self, qn_auth):
         assert resp.status_code == 200, resp
         assert ret is not None, resp
 
-    def test_pfop_idle_time_task(self, set_conf_default, qn_auth):
-        persistence_key = 'python-sdk-pfop-test/test-pfop-by-api'
+    @pytest.mark.parametrize(
+        'persistent_options',
+        (
+            # included by above test_pfop_execute
+            # {
+            #     'persistent_type': None,
+            # },
+            {
+                'persistent_type': 0,
+            },
+            {
+                'persistent_type': 1,
+            },
+            {
+                'workflow_template_id': 'test-workflow',
+            },
+        )
+    )
+    def test_pfop_idle_time_task(
+        self,
+        set_conf_default,
+        qn_auth,
+        bucket_name,
+        persistent_options,
+    ):
+        persistent_type = persistent_options.get('persistent_type')
+        workflow_template_id = persistent_options.get('workflow_template_id', None)
+
+        execute_opts = {}
+        if workflow_template_id:
+            execute_opts['workflow_template_id'] = workflow_template_id
+        else:
+            persistent_key = '_'.join([
+                'test-pfop/test-pfop-by-api',
+                'type',
+                str(persistent_type)
+            ])
+            execute_opts['fops'] = [
+                op_save(
+                    op='avinfo',
+                    bucket=bucket_name,
+                    key=persistent_key
+                )
+            ]
+
+        if persistent_type is not None:
+            execute_opts['persistent_type'] = persistent_type
+
+        pfop = PersistentFop(qn_auth, bucket_name)
+        key = 'qiniu.png'
+        ret, resp = pfop.execute(
+            key,
+            **execute_opts
+        )
 
-        key = 'sintel_trailer.mp4'
-        pfop = PersistentFop(qn_auth, 'testres')
-        ops = [
-            op_save(
-                op='avthumb/m3u8/segtime/10/vcodec/libx264/s/320x240',
-                bucket='pythonsdk',
-                key=persistence_key
-            )
-        ]
-        ret, resp = pfop.execute(key, ops, force=1, persistent_type=1)
         assert resp.status_code == 200, resp
+        assert ret is not None
         assert 'persistentId' in ret, resp
 
         ret, resp = pfop.get_status(ret['persistentId'])
         assert resp.status_code == 200, resp
-        assert ret['type'] == 1, resp
+        assert ret is not None
         assert ret['creationDate'] is not None, resp
+
+        if persistent_id == 1:
+            assert ret['type'] == 1, resp
+        elif workflow_template_id:
+            assert workflow_template_id in ret['taskFrom'], resp

tests/cases/test_services/test_storage/test_upload_pfop.py

@@ -12,36 +12,60 @@
 # or this test will continue to occupy bucket space.
 class TestPersistentFopByUpload:
     @pytest.mark.parametrize('temp_file', [10 * MB], indirect=True)
-    @pytest.mark.parametrize('persistent_type', [None, 0, 1])
+    @pytest.mark.parametrize(
+        'persistent_options',
+        (
+            {
+                'persistent_type': None,
+            },
+            {
+                'persistent_type': 0,
+            },
+            {
+                'persistent_type': 1,
+            },
+            {
+                'persistent_workflow_template_id': 'test-workflow',
+            },
+        )
+    )
     def test_pfop_with_upload(
         self,
         set_conf_default,
         qn_auth,
         bucket_name,
         temp_file,
-        persistent_type
+        persistent_options,
     ):
-        key = 'test-pfop-upload-file'
-        persistent_key = '_'.join([
-            'test-pfop-by-upload',
-            'type',
-            str(persistent_type)
-        ])
-        persistent_ops = ';'.join([
-            qiniu.op_save(
-                op='avthumb/m3u8/segtime/10/vcodec/libx264/s/320x240',
-                bucket=bucket_name,
-                key=persistent_key
-            )
-        ])
+        key = 'test-pfop/upload-file'
+        persistent_type = persistent_options.get('persistent_type')
+        persistent_workflow_template_id = persistent_options.get('persistent_workflow_template_id')
+
+        upload_policy = {}
 
-        upload_policy = {
-            'persistentOps': persistent_ops
-        }
+        # set pfops or tmplate id
+        if persistent_workflow_template_id:
+            upload_policy['persistentWorkflowTemplateID'] = persistent_workflow_template_id
+        else:
+            persistent_key = '_'.join([
+                'test-pfop/test-pfop-by-upload',
+                'type',
+                str(persistent_type)
+            ])
+            persistent_ops = ';'.join([
+                qiniu.op_save(
+                    op='avinfo',
+                    bucket=bucket_name,
+                    key=persistent_key
+                )
+            ])
+            upload_policy['persistentOps'] = persistent_ops
 
+        # set persistent type
         if persistent_type is not None:
             upload_policy['persistentType'] = persistent_type
 
+        # upload
         token = qn_auth.upload_token(
             bucket_name,
             key,
@@ -61,6 +85,10 @@ def test_pfop_with_upload(
         pfop = qiniu.PersistentFop(qn_auth, bucket_name)
         ret, resp = pfop.get_status(ret['persistentId'])
         assert resp.status_code == 200, resp
+        assert ret is not None, resp
+        assert ret['creationDate'] is not None, resp
+
         if persistent_type == 1:
             assert ret['type'] == 1, resp
-        assert ret['creationDate'] is not None, resp
+        elif persistent_workflow_template_id:
+            assert persistent_workflow_template_id in ret['taskFrom'], resp