Manage resource pool and bucket QoS with OSS SDK for Python - Object Storage Service

Use the OSS SDK for Python V2 to configure, query, and delete Quality of Service (QoS) throttling rules at three levels: bucket, requester, and bucket group.

How it works

Resource pool QoS lets you proactively set bandwidth ceilings rather than relying on the platform to rate-limit traffic reactively. Rules cascade across three management levels:

Bucket level: Set a bandwidth ceiling for a bucket within its resource pool.
Requester level: Set per-requester bandwidth limits for a specific bucket or resource pool.
Bucket group level: Group buckets within a resource pool and apply shared bandwidth limits to each group.

All examples use oss.OperationInput to call the underlying API directly and load credentials from environment variables.

Operations reference

The following table lists all operations covered in this topic.

Section	Operation	Method	Key parameter
Bucket bandwidth management	`PutBucketQoSInfo`	PUT	`qosInfo`
Bucket bandwidth management	`GetBucketQoSInfo`	GET	`qosInfo`
Bucket bandwidth management	`DeleteBucketQoSInfo`	DELETE	`qosInfo`
Bucket-level bandwidth management for requesters	`PutBucketRequesterQoSInfo`	PUT	`requesterQosInfo`
Bucket-level bandwidth management for requesters	`GetBucketRequesterQoSInfo`	GET	`requesterQosInfo`
Bucket-level bandwidth management for requesters	`ListBucketRequesterQoSInfos`	GET	`requesterQosInfo`
Bucket-level bandwidth management for requesters	`DeleteBucketRequesterQoSInfo`	DELETE	`requesterQosInfo`
Resource pool-level bandwidth management	`ListResourcePools`	GET	`resourcePool`
Resource pool-level bandwidth management	`GetResourcePoolInfo`	GET	`resourcePoolInfo`
Resource pool-level bandwidth management	`ListResourcePoolBucketGroups`	GET	`resourcePoolBucketGroup`
Resource pool-level bandwidth management	`PutResourcePoolRequesterQoSInfo`	PUT	`requesterQosInfo`
Resource pool-level bandwidth management	`GetResourcePoolRequesterQoSInfo`	GET	`requesterQosInfo`
Resource pool-level bandwidth management	`ListResourcePoolRequesterQoSInfos`	GET	`requesterQosInfo`
Resource pool-level bandwidth management	`DeleteResourcePoolRequesterQoSInfo`	DELETE	`requesterQosInfo`
Bucket group bandwidth management	`PutBucketResourcePoolBucketGroup`	PUT	`resourcePoolBucketGroup`
Bucket group bandwidth management	`ListResourcePoolBucketGroups`	GET	`resourcePoolBucketGroup`
Bucket group bandwidth management	`PutResourcePoolBucketGroupQoSInfo`	PUT	`resourcePoolBucketGroupQosInfo`
Bucket group bandwidth management	`GetResourcePoolBucketGroupQoSInfo`	GET	`resourcePoolBucketGroupQosInfo`
Bucket group bandwidth management	`ListResourcePoolBucketGroupQoSInfos`	GET	`resourcePoolBucketGroupQosInfo`
Bucket group bandwidth management	`DeleteResourcePoolBucketGroupQoSInfo`	DELETE	`resourcePoolBucketGroupQosInfo`

Usage notes

The examples use the region ID cn-hangzhou. Replace it with the region where your bucket is located.
By default, examples use a public endpoint. To access OSS from another Alibaba Cloud service in the same region, use an internal endpoint instead. For more information, see OSS regions and endpoints.

Prerequisites

Before you begin, ensure that you have:

An OSS bucket in an existing resource pool
Access credentials stored as environment variables (OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
A qos.xml file containing the QoS configuration body (required for write operations)

Bucket bandwidth management

Use these operations to set, query, and delete bandwidth throttling rules for a bucket within its resource pool.

Set throttling rules for a bucket

import alibabacloud_oss_v2 as oss

def PutBucketQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    with open('qos.xml', 'r') as f:
        qos_xml_body = f.read()

    req = oss.OperationInput(
        op_name='PutBucketQoSInfo',
        method='PUT',
        parameters={'qosInfo': ''},
        headers=None,
        body=qos_xml_body,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)

if __name__ == "__main__":
    PutBucketQoSInfo()

Get throttling rules for a bucket

import alibabacloud_oss_v2 as oss

def GetBucketQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='GetBucketQoSInfo',
        method='GET',
        parameters={'qosInfo': ''},
        headers=None,
        body=None,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    GetBucketQoSInfo()

Delete throttling rules for a bucket

import alibabacloud_oss_v2 as oss

def DeleteBucketQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='DeleteBucketQoSInfo',
        method='DELETE',
        parameters={'qosInfo': ''},
        headers=None,
        body=None,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)

if __name__ == "__main__":
    DeleteBucketQoSInfo()

Bucket-level bandwidth management for requesters

Use these operations to set per-requester bandwidth limits for a specific bucket. The requester is identified by an Alibaba Cloud account UID passed as the qosRequester parameter.

Set throttling rules for a requester accessing a bucket

import alibabacloud_oss_v2 as oss

def PutBucketRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    with open('qos.xml', 'r') as f:
        qos_xml_body = f.read()

    req = oss.OperationInput(
        op_name='PutBucketRequesterQoSInfo',
        method='PUT',
        parameters={
            'requesterQosInfo': '',
            'qosRequester': '2598732222222xxxx',  # Requester account UID
        },
        headers=None,
        body=qos_xml_body,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    PutBucketRequesterQoSInfo()

Get throttling rules for a requester accessing a bucket

import alibabacloud_oss_v2 as oss

def GetBucketRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='GetBucketRequesterQoSInfo',
        method='GET',
        parameters={
            'requesterQosInfo': '',
            'qosRequester': '2598732222222xxxx',
        },
        headers=None,
        body=None,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    GetBucketRequesterQoSInfo()

List throttling rules for all requesters accessing a bucket

Supports pagination via continuation-token and max-keys.

import alibabacloud_oss_v2 as oss

def ListBucketRequesterQoSInfos():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListBucketRequesterQoSInfos',
        method='GET',
        parameters={
            'requesterQosInfo': '',
            # 'continuation-token': '2345',  # Resume a paged query from this token
            # 'max-keys': '1',              # Limit the number of results returned
        },
        headers=None,
        body=None,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    ListBucketRequesterQoSInfos()

Delete throttling rules for a requester accessing a bucket

import alibabacloud_oss_v2 as oss

def DeleteBucketRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='DeleteBucketRequesterQoSInfo',
        method='DELETE',
        parameters={
            'requesterQosInfo': '',
            'qosRequester': '2598732222222xxxx',
        },
        headers=None,
        body=None,
        bucket='examplebucket',
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    DeleteBucketRequesterQoSInfo()

Resource pool-level bandwidth management

Use these operations to query resource pools and manage per-requester bandwidth limits at the resource pool level. Resource pool operations do not require a bucket parameter.

List all resource pools in the account

Supports pagination via continuation-token and max-keys.

import alibabacloud_oss_v2 as oss

def ListResourcePools():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListResourcePools',
        method='GET',
        parameters={
            'resourcePool': '',
            # 'continuation-token': 'test-rp-',
            # 'max-keys': '1',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == '__main__':
    ListResourcePools()

Get information about a resource pool

import alibabacloud_oss_v2 as oss

def GetResourcePoolInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='GetResourcePoolInfo',
        method='GET',
        parameters={
            'resourcePoolInfo': '',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    GetResourcePoolInfo()

List bucket groups in a resource pool

import alibabacloud_oss_v2 as oss

def ListResourcePoolBucketGroups():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListResourcePoolBucketGroups',
        method='GET',
        parameters={
            'resourcePoolBucketGroup': '',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    ListResourcePoolBucketGroups()

Set throttling rules for a requester in a resource pool

import alibabacloud_oss_v2 as oss

def PutResourcePoolRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    with open('qos.xml', 'r') as f:
        qos_xml_body = f.read()

    req = oss.OperationInput(
        op_name='PutResourcePoolRequesterQoSInfo',
        method='PUT',
        parameters={
            'requesterQosInfo': '',
            'resourcePool': 'example-resource-pool',
            'qosRequester': '2598732222222xxxx',
        },
        headers=None,
        body=qos_xml_body,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    PutResourcePoolRequesterQoSInfo()

Get throttling rules for a requester in a resource pool

import alibabacloud_oss_v2 as oss

def GetResourcePoolRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='GetResourcePoolRequesterQoSInfo',
        method='GET',
        parameters={
            'requesterQosInfo': '',
            'qosRequester': '2598732222222xxxx',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    GetResourcePoolRequesterQoSInfo()

List throttling rules for all requesters in a resource pool

Supports pagination via continuation-token and max-keys.

import alibabacloud_oss_v2 as oss

def ListResourcePoolRequesterQoSInfos():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListResourcePoolRequesterQoSInfos',
        method='GET',
        parameters={
            'requesterQosInfo': '',
            'resourcePool': 'example-resource-pool',
            # 'continuation-token': '2345',
            # 'max-keys': '1',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    ListResourcePoolRequesterQoSInfos()

Delete throttling rules for a requester in a resource pool

import alibabacloud_oss_v2 as oss

def DeleteResourcePoolRequesterQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='DeleteResourcePoolRequesterQoSInfo',
        method='DELETE',
        parameters={
            'requesterQosInfo': '',
            'resourcePool': 'example-resource-pool',
            'qosRequester': '2598732222222xxxx',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    DeleteResourcePoolRequesterQoSInfo()

Bucket group bandwidth management

Bucket groups let you apply shared bandwidth limits to a subset of buckets within a resource pool. Add buckets to a group first, then configure QoS rules for the group.

Add a bucket to a bucket group

import alibabacloud_oss_v2 as oss

def PutBucketResourcePoolBucketGroup():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='PutBucketResourcePoolBucketGroup',
        method='PUT',
        parameters={
            'resourcePoolBucketGroup': 'example-group',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket='examplebucket',  # The bucket to add to the group
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    PutBucketResourcePoolBucketGroup()

List bucket groups in a resource pool

import alibabacloud_oss_v2 as oss

def ListResourcePoolBucketGroups():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListResourcePoolBucketGroups',
        method='GET',
        parameters={
            'resourcePoolBucketGroup': '',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    ListResourcePoolBucketGroups()

Set throttling rules for a bucket group

import alibabacloud_oss_v2 as oss

def PutResourcePoolBucketGroupQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    with open('qos.xml', 'r') as f:
        qos_xml_body = f.read()

    req = oss.OperationInput(
        op_name='PutResourcePoolBucketGroupQoSInfo',
        method='PUT',
        parameters={
            'resourcePoolBucketGroupQosInfo': '',
            'resourcePool': 'example-resource-pool',
            'resourcePoolBucketGroup': 'example-group',
        },
        headers=None,
        body=qos_xml_body,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    PutResourcePoolBucketGroupQoSInfo()

Get throttling rules for a bucket group

import alibabacloud_oss_v2 as oss

def GetResourcePoolBucketGroupQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='GetResourcePoolBucketGroupQoSInfo',
        method='GET',
        parameters={
            'resourcePoolBucketGroupQosInfo': '',
            'resourcePool': 'example-resource-pool',
            'resourcePoolBucketGroup': 'example-group',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    GetResourcePoolBucketGroupQoSInfo()

List throttling rules for all bucket groups in a resource pool

import alibabacloud_oss_v2 as oss

def ListResourcePoolBucketGroupQoSInfos():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='ListResourcePoolBucketGroupQoSInfos',
        method='GET',
        parameters={
            'resourcePoolBucketGroupQosInfo': '',
            'resourcePool': 'example-resource-pool',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    ListResourcePoolBucketGroupQoSInfos()

Delete throttling rules for a bucket group

import alibabacloud_oss_v2 as oss

def DeleteResourcePoolBucketGroupQoSInfo():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = "cn-hangzhou"
    client = oss.Client(cfg)

    req = oss.OperationInput(
        op_name='DeleteResourcePoolBucketGroupQoSInfo',
        method='DELETE',
        parameters={
            'resourcePoolBucketGroupQosInfo': '',
            'resourcePool': 'example-resource-pool',
            'resourcePoolBucketGroup': 'example-group',
        },
        headers=None,
        body=None,
        bucket=None,
    )

    resp = client.invoke_operation(req)
    print(resp.status_code)
    print(resp.headers)
    print(resp.http_response.content)

if __name__ == "__main__":
    DeleteResourcePoolBucketGroupQoSInfo()

What's next

Operations related to resource pool QoS — Full API reference for all QoS operations covered in this topic.
Examples of resource pool QoS configuration — End-to-end configuration examples showing how to set up QoS policies in practice.