使用PythonSDK對管道進行搜尋、查詢、更新 - ApsaraVideo Media Processing

管道是任務的隊列。當您提交非同步任務後，任務會進入管道中進行排隊，根據優先順序和提交順序依次被調度執行。本文提供了Python SDK V2.0管道相關功能的API調用樣本，包含添加管道、更新管道、刪除管道、查詢管道。

搜尋管道

可以直接通過SearchPipeline介面搜尋管道資訊。

import os
import sys

from typing import List

from alibabacloud_mts20140618.client import Client as Mts20140618Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_mts20140618 import models as mts_20140618_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Mts20140618Client:
        """
        使用AK&SK初始化帳號Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'mts.cn-hangzhou.aliyuncs.com'
        return Mts20140618Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        search_pipeline_request = mts_20140618_models.SearchPipelineRequest(
            #需要搜尋的管道狀態
            state='Paused',
            #分頁查詢時設定的每頁行數大小
            page_size=10,
            #當前頁號
            page_number=1
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            client.search_pipeline_with_options(search_pipeline_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        search_pipeline_request = mts_20140618_models.SearchPipelineRequest(
            # 需要搜尋的管道狀態
            state='Paused',
            # 分頁查詢時設定的每頁行數大小
            page_size=10,
            # 當前頁號
            page_number=1
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            await client.search_pipeline_with_options_async(search_pipeline_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

查詢管道

如果已經知道pipelineId，可以通過pipelineId調用QueryPipelineList介面查詢管道資訊。

import os
import sys

from typing import List

from alibabacloud_mts20140618.client import Client as Mts20140618Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_mts20140618 import models as mts_20140618_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Mts20140618Client:
        """
        使用AK&SK初始化帳號Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'mts.cn-hangzhou.aliyuncs.com'
        return Mts20140618Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        query_pipeline_list_request = mts_20140618_models.QueryPipelineListRequest(
            #需要查詢的管道ID列表
            pipeline_ids='d1ce4d3efcb549419193f50f1fcd****,72dfa5e679ab4be9a3ed9974c736****'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            client.query_pipeline_list_with_options(query_pipeline_list_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        query_pipeline_list_request = mts_20140618_models.QueryPipelineListRequest(
            # 需要查詢的管道ID列表
            pipeline_ids='d1ce4d3efcb549419193f50f1fcd****,72dfa5e679ab4be9a3ed9974c736****'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            await client.query_pipeline_list_with_options_async(query_pipeline_list_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

更新管道

通過UpdatePipeline更新管道資訊，包括更新管道名稱，狀態。管道的狀態包括Active、Paused。

import os
import sys

from typing import List

from alibabacloud_mts20140618.client import Client as Mts20140618Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_mts20140618 import models as mts_20140618_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Mts20140618Client:
        """
        使用AK&SK初始化帳號Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'mts.cn-hangzhou.aliyuncs.com'
        return Mts20140618Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        update_pipeline_request = mts_20140618_models.UpdatePipelineRequest(
            #需要更新的管道ID
            pipeline_id='d1ce4d3efcb549419193f50f1fcd****',
            #修改後的管道名稱
            name='example-pipeline-****',
            #修改後的管道狀態
            state='Paused',
            #阿里雲Message Service配置
            notify_config='{"Topic":"example-topic-****"}',
            #當前RAM使用者關聯的角色
            role='AliyunMTSDefaultRole'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            client.update_pipeline_with_options(update_pipeline_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        update_pipeline_request = mts_20140618_models.UpdatePipelineRequest(
            pipeline_id='d1ce4d3efcb549419193f50f1fcd****',
            name='example-pipeline-****',
            state='Paused',
            notify_config='{"Topic":"example-topic-****"}',
            role='AliyunMTSDefaultRole'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 複製代碼運行請自行列印 API 的傳回值
            await client.update_pipeline_with_options_async(update_pipeline_request, runtime)
        except Exception as error:
            # 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            # 錯誤 message
            print(error.message)
            # 診斷地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

ApsaraVideo Media Processing：管道管理

搜尋管道

查詢管道

更新管道

相關文檔