Python V2 OSS SDK での同期処理の実装 - Object Storage Service - Alibaba Cloud ドキュメントセンター

非同期処理では、プログラムはタスクが完了するのを待たずに他のタスクの実行を開始できます。Object Storage Service (OSS) でデータを非同期に処理するには、x-oss-async-process パラメーターを使用します。このトピックでは、ドキュメント変換、ビデオトランスコード、ビデオマージなどの非同期処理シナリオで Python V2 用 OSS SDK を使用する方法について説明します。

使用上の注意

このトピックのサンプルコードでは、中国 (杭州) リージョンのリージョン ID cn-hangzhou を使用しています。デフォルトでは、パブリックエンドポイントを使用してバケット内のリソースにアクセスします。バケットが配置されているのと同じリージョン内の他の Alibaba Cloud サービスを使用してバケット内のリソースにアクセスする場合は、内部エンドポイントを使用します。OSS のリージョンとエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。

メソッド定義

async_process_object(request: AsyncProcessObjectRequest, **kwargs) → AsyncProcessObjectResult

リクエストパラメーター

パラメーター	タイプ	説明
request	AsyncProcessObjectRequest	AsyncProcessObject 操作のリクエスト。リクエストのパラメーターについては、「AsyncProcessObjectRequest」を参照してください。

レスポンスパラメーター

タイプ	説明
AsyncProcessObjectResult	戻り値。詳細については、「AsyncProcessObjectResult」を参照してください。

完全なメソッド定義については、「async_process_object」をご参照ください。

サンプルコード

次のサンプルコードは、Python V2 用 OSS SDK を使用してドキュメントのフォーマットを変換します。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、デスティネーションオブジェクト名、デスティネーションバケット名を含む、コマンドライン引数を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)

def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # DOCX を PNG に変換するためのスタイルを定義します。
    style = "doc/convert,target_png,source_docx"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名。
        key=args.key,  # ソースオブジェクト名。
        process=process, # 処理命令。
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )

# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

一般的なシナリオ

ビデオトランスコード

ビデオトランスコードを使用すると、ビデオのエンコード形式とコンテナ形式を変更したり、解像度やビットレートなどのパラメーターを調整することでビデオファイルサイズを縮小したりできます。次のサンプルコードは、ビデオをトランスコードします。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、変換後のオブジェクト名、および変換後のオブジェクトを格納するバケット名を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)


def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # ビデオ処理のスタイルを定義します。この例では、ビデオトランスコード設定には、コンテナ形式、エンコード形式、解像度、ビデオビットレート、フレームレート、オーディオコーデック、およびオーディオビットレートが含まれます。
    style = "video/convert,f_avi,vcodec_h265,s_1920x1080,vb_2000000,fps_30,acodec_aac,ab_100000,sn_1"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名
        key=args.key,  # ソースオブジェクト名
        process=process, # 処理命令
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )


# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

ビデオをアニメーション画像に変換する

ビデオからアニメーション画像への変換は、ビデオを GIF または WebP 形式のアニメーション画像に変換するメディア処理技術です。次のサンプルコードは、ビデオからアニメーション画像を作成します。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、デスティネーションオブジェクト名、デスティネーションバケット名を含む、コマンドライン引数を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)

def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # ビデオをアニメーション画像に変換するためのスタイルを定義します。この例では、設定には GIF の幅、GIF の高さ、およびフレームをキャプチャする時間間隔が含まれます。
    style = "video/animation,f_gif,w_100,h_100,inter_1000"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名。
        key=args.key,  # ソースオブジェクト名。
        process=process, # 処理命令。
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )

# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

フレームキャプチャ

フレームキャプチャを使用して、要件に基づいてビデオフレームをキャプチャし、特定の形式に変換できます。次のサンプルコードは、ビデオからフレームをキャプチャします。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、デスティネーションオブジェクト名、デスティネーションバケット名を含む、コマンドライン引数を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)

def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # フレームキャプチャのスタイルを定義します。
    style = "video/snapshots,f_jpg,w_100,h_100,scaletype_crop,inter_10000"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名。
        key=args.key,  # ソースオブジェクト名。
        process=process, # 処理命令。
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )

# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

オーディオトランスコード

オーディオトランスコードを使用すると、オーディオファイルをある形式から別の形式に変換できます。次のサンプルコードは、オーディオファイルの形式を指定された形式に変換します。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、デスティネーションオブジェクト名、デスティネーションバケット名を含む、コマンドライン引数を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)


def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # オーディオトランスコードのスタイルを定義します。
    style = "audio/convert,ss_10000,t_60000,f_aac,ab_96000"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名。
        key=args.key,  # ソースオブジェクト名。
        process=process, # 処理命令。
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )


# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

不可視ウォーターマークのデコード

次のサンプルコードは、画像から不可視ウォーターマークをデコードします。

import base64
import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数を解析するためのパーサーを作成し、スクリプトの目的を記述します。
parser = argparse.ArgumentParser(description="async process object sample")

# 必須のリージョン、ソースバケット名、エンドポイント、ソースオブジェクト名、デスティネーションオブジェクト名、デスティネーションバケット名を含む、コマンドライン引数を指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--target_key', help='処理済みオブジェクトの名前を指定します。', required=True)
parser.add_argument('--target_bucket', help='処理済みオブジェクトの格納に使用するバケットの名前を指定します。', required=True)


def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 認証のために環境変数からアクセス認証情報を取得します。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルト構成を使用して cfg オブジェクトを作成し、認証情報プロバイダーを指定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # cfg オブジェクトのリージョン属性を、コマンドラインで指定されたリージョンに設定します。
    cfg.region = args.region

    # カスタムエンドポイントが指定されている場合は、cfg オブジェクトのエンドポイント属性を指定されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 前述の設定を使用して、OSSClient インスタンスを初期化します。
    client = oss.Client(cfg)

    # 指定された画像からウォーターマークをデコードします。
    style = "image/deblindwatermark,s_low,t_text"

    # デスティネーションバケット名とオブジェクト名を Base64 エンコードして、URL セーフにします。
    target_bucket_base64 = base64.b64encode(args.target_bucket.encode()).decode()
    target_key_base64 = base64.b64encode(args.target_key.encode()).decode()

    # スタイルとデスティネーションパスを含む処理命令を作成します。
    process = f"{style}|sys/saveas,o_{target_key_base64},b_{target_bucket_base64}"

    # オブジェクトを非同期に処理し、結果のオブジェクトをデスティネーションバケットに格納するリクエストを送信します。
    result = client.async_process_object(oss.AsyncProcessObjectRequest(
        bucket=args.bucket,  # ソースバケット名。
        key=args.key,  # ソースオブジェクト名。
        process=process, # 処理命令。
    ))

    # リクエストのステータスと結果を確認するために、ステータスコードとその他の結果情報を表示します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' event id: {result.event_id},'
          f' task id: {result.task_id},'
          f' process request id: {result.process_request_id},'
          )


# スクリプトが直接実行されたときに処理ロジックを開始するために、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイントを指定します。制御フローはここから開始されます。

Object Storage Service:Python 2.0 OSS SDK を使用した非同期処理