Python用のサーバーSDKを使用する - ApsaraVideo Live - Alibaba Cloud ドキュメントセンター

このトピックでは、ApsaraVideo Liveが提供するサーバーSDK for Pythonの使用方法と、関連するサンプルコードについて説明します。このトピックでは、ApsaraVideo Live API操作を呼び出す方法を示す例として、プロダクションスタジオをクエリするAPI操作を使用します。

前提条件

Python 2.7以降がインストールされます。
Python用のサーバーSDKがダウンロードされます。詳細については、「SDKダウンロード」をご参照ください。

手順

SDKをインストールします。
1. Alibaba Cloud Core SDKをインストールします。
  sudo pip install aliyun-python-sdk-core
2. ApsaraVideo Live SDKをインストールします。
  sudo pip install aliyun-python-sdk-live
SDKを更新します。
1. Alibaba Cloud Core SDKを更新します。
  sudo pip install aliyun-python-sdk-core --upgrade
2. ApsaraVideo Live SDKを更新します。
  sudo pip install aliyun-python-sdk-live --upgrade

PythonのサーバーSDKを使用します。

ファイルの説明を読み取ります。この例では、SDKのDescribeCastersRequest.pyファイルが使用されています。

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from aliyunsdkcore.request import RpcRequest
from aliyunsdklive.endpoint import endpoint_data

class DescribeCastersRequest(RpcRequest):

    def __init__(self):
        RpcRequest.__init__(self, 'live', '2016-11-01', 'DescribeCasters','live')
        self.set_method('POST')

        if hasattr(self, "endpoint_map"):
            setattr(self, "endpoint_map", endpoint_data.getEndpointMap())
        if hasattr(self, "endpoint_regional"):
            setattr(self, "endpoint_regional", endpoint_data.getEndpointRegional())

    def get_StartTime(self): # String
        return self.get_query_params().get('StartTime')

    def set_StartTime(self, StartTime):  # String
        self.add_query_param('StartTime', StartTime)
    def get_PageNum(self): # Integer
        return self.get_query_params().get('PageNum')

    def set_PageNum(self, PageNum):  # Integer
        self.add_query_param('PageNum', PageNum)
    def get_CasterName(self): # String
        return self.get_query_params().get('CasterName')

    def set_CasterName(self, CasterName):  # String
        self.add_query_param('CasterName', CasterName)
    def get_PageSize(self): # Integer
        return self.get_query_params().get('PageSize')

    def set_PageSize(self, PageSize):  # Integer
        self.add_query_param('PageSize', PageSize)
    def get_NormType(self): # String
        return self.get_query_params().get('NormType')

    def set_NormType(self, NormType):  # String
        self.add_query_param('NormType', NormType)
    def get_CasterId(self): # String
        return self.get_query_params().get('CasterId')

    def set_CasterId(self, CasterId):  # String
        self.add_query_param('CasterId', CasterId)
    def get_EndTime(self): # String
        return self.get_query_params().get('EndTime')

    def set_EndTime(self, EndTime):  # String
        self.add_query_param('EndTime', EndTime)
    def get_OwnerId(self): # Long
        return self.get_query_params().get('OwnerId')

    def set_OwnerId(self, OwnerId):  # Long
        self.add_query_param('OwnerId', OwnerId)
    def get_OrderByModifyAsc(self): # String
        return self.get_query_params().get('OrderByModifyAsc')

    def set_OrderByModifyAsc(self, OrderByModifyAsc):  # String
        self.add_query_param('OrderByModifyAsc', OrderByModifyAsc)
    def get_ChargeType(self): # Integer
        return self.get_query_params().get('ChargeType')

        def set_ChargeType(self, ChargeType):  # Integer
        self.add_query_param('ChargeType', ChargeType)
        def get_Status(self): # Integer
        return self.get_query_params().get('Status')

        def set_Status(self, Status):  # Integer
        self.add_query_param('Status', Status)

ファイル内のDescribeCastersRequestクラスには、対応するget_Xメソッドとset_Xメソッドの複数のセットがあり、それぞれAPIリクエストのパラメーターの取得と指定に使用されます。

config.iniという名前の構成ファイルを作成し、confディレクトリに配置します。 AccessKey IDとAccessKeyシークレットを設定ファイルに含めます。例:

[default]
access_key_id = YOUR_ACCESS_KEY_ID
access_key_secret = YOUR_ACCESS_KEY_SECRET

YOUR_ACCESS_KEY_IDとYOUR_ACCESS_KEY_SECRETを実際のAccessKey IDとAccessKeyシークレットに置き換えます。

次に、次のPythonコードを使用して構成ファイルを読み取り、SDKを呼び出すことができます。

次のサンプルコードは、SDKのv20161101でDescribeCasters操作を例として使用しています。

#!/usr/bin/env python
#coding=utf-8

from configparser import ConfigParser
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.auth.credentials import AccessKeyCredential
from aliyunsdkcore.auth.credentials import StsTokenCredential
from aliyunsdklive.request.v20161101.DescribeCastersRequest import DescribeCastersRequest

config = ConfigParser()
config.read('conf/config.ini')

# The AccessKey pair of an Alibaba Cloud account has access permissions on all API operations. We recommend that you use the AccessKey pair of a RAM user to call API operations or perform routine O&M. 
# We recommend that you not save your AccessKey pair (AccessKey ID and AccessKey secret) in your project code. Otherwise, the AccessKey pair may be leaked and the security of all resources within your account may be compromised. 
# In this example, the AccessKey pair is obtained from the configuration file to authenticate API accesses. 
access_key_id = config.get('default', 'access_key_id')
access_key_secret = config.get('default', 'access_key_secret')
# Initialize the client.
# Use the AccessKey ID and AccessKey secret of your Alibaba Cloud account. You can obtain them in the Alibaba Cloud Management Console.
# Set the region to cn-hangzhou in this example.
credentials = AccessKeyCredential(access_key_id, access_key_secret)
# use STS Token
# credentials = StsTokenCredential('<your-access-key-id>', '<your-access-key-secret>', '<your-sts-token>')
client = AcsClient(region_id='cn-hangzhou', credential=credentials)
# Construct an API request.
request = DescribeCastersRequest()
request.set_accept_format('json')

request.set_CasterId("testCase")
request.set_CasterName("testName")
request.set_StartTime("2020-01-01T12:00:01")
# Send the request.
response = client.do_action_with_exception(request)
# python2:  print(response)
# Obtain the result.
print(str(response, encoding='utf-8'))

次のステップ

関連するSDKをアンインストールするには、次のコマンドを実行します。

sudo pip uninstall aliyun-python-sdk-core
sudo pip uninstall aliyun-python-sdk-live