Object Storage Service:Bucket inventory (Python SDK V2)

Prerequisites

Before you begin, make sure that you have:

• Installed the alibabacloud_oss_v2 package

• An OSS bucket with the required permissions (create, query, list, and delete bucket inventories). By default, the bucket owner has these permissions. If you don't have them, ask the bucket owner to grant access.

• A RAM role whose Alibaba Cloud Resource Name (ARN) follows the format acs:ram::<uid>:role/<rolename>, with read access to the source bucket and write access to the destination bucket

Usage notes

• All examples use the cn-hangzhou region (China (Hangzhou)) with a public endpoint. To access OSS from other Alibaba Cloud services in the same region, use an internal endpoint. For endpoint details, see Regions and endpoints.

• Each bucket supports up to 1,000 inventory configurations.

• The source bucket and destination bucket must be in the same region.

• A single list request returns up to 100 inventory configurations. Use the continuation token from each response to retrieve the next page.

Client initialization

All examples use the same client initialization pattern: credentials are read from environment variables, and the client is built from a configuration object.

import alibabacloud_oss_v2 as oss

credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
cfg.region = "<region>"          # e.g., cn-hangzhou
# cfg.endpoint = "<endpoint>"    # Set only if using a custom or internal endpoint

client = oss.Client(cfg)

Create an inventory configuration

client.put_bucket_inventory(oss.PutBucketInventoryRequest(...))

Required parameters

Optional parameters

Example

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket inventory sample")
parser.add_argument('--region', required=True)
parser.add_argument('--bucket', required=True)
parser.add_argument('--endpoint')
parser.add_argument('--user_id', required=True)
parser.add_argument('--arn', required=True)
parser.add_argument('--inventory_id', required=True)

def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    result = client.put_bucket_inventory(oss.PutBucketInventoryRequest(
        bucket=args.bucket,
        inventory_id=args.inventory_id,
        inventory_configuration=oss.InventoryConfiguration(
            id=args.inventory_id,
            is_enabled=True,
            included_object_versions='All',
            optional_fields=oss.OptionalFields(
                fields=[
                    oss.InventoryOptionalFieldType.SIZE,
                    oss.InventoryOptionalFieldType.LAST_MODIFIED_DATE,
                ],
            ),
            destination=oss.InventoryDestination(
                oss_bucket_destination=oss.InventoryOSSBucketDestination(
                    format=oss.InventoryFormatType.CSV,
                    account_id=args.user_id,
                    role_arn=args.arn,
                    bucket=f'acs:oss:::{args.bucket}',
                    prefix='aaa',
                ),
            ),
            schedule=oss.InventorySchedule(
                frequency=oss.InventoryFrequencyType.DAILY,
            ),
            filter=oss.InventoryFilter(
                lower_size_bound=1024,
                upper_size_bound=1048576,
                storage_class='ColdArchive',
                prefix='aaa',
                last_modify_begin_time_stamp=1637883649,
                last_modify_end_time_stamp=1638347592,
            ),
        ),
    ))

    print(f'status code: {result.status_code}, request id: {result.request_id}')

if __name__ == "__main__":
    main()

Query an inventory configuration

client.get_bucket_inventory(oss.GetBucketInventoryRequest(...))

Required parameters

Response fields

result.status_code
result.request_id
result.inventory_configuration.id
result.inventory_configuration.is_enabled
result.inventory_configuration.included_object_versions
result.inventory_configuration.destination.oss_bucket_destination.account_id
result.inventory_configuration.destination.oss_bucket_destination.role_arn
result.inventory_configuration.destination.oss_bucket_destination.bucket
result.inventory_configuration.destination.oss_bucket_destination.prefix
# result.inventory_configuration.destination.oss_bucket_destination.encryption.sse_kms.key_id
# result.inventory_configuration.destination.oss_bucket_destination.encryption.sse_oss
result.inventory_configuration.filter.lower_size_bound
result.inventory_configuration.filter.upper_size_bound
result.inventory_configuration.filter.storage_class
result.inventory_configuration.filter.prefix
result.inventory_configuration.filter.last_modify_begin_time_stamp
result.inventory_configuration.filter.last_modify_end_time_stamp

Example

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="get bucket inventory sample")
parser.add_argument('--region', required=True)
parser.add_argument('--bucket', required=True)
parser.add_argument('--endpoint')
parser.add_argument('--inventory_id', required=True)

def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    result = client.get_bucket_inventory(oss.GetBucketInventoryRequest(
        bucket=args.bucket,
        inventory_id=args.inventory_id,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' included object versions: {result.inventory_configuration.included_object_versions},'
          f' id: {result.inventory_configuration.id},'
          f' is enabled: {result.inventory_configuration.is_enabled},'
          f' account id: {result.inventory_configuration.destination.oss_bucket_destination.account_id},'
          f' role arn: {result.inventory_configuration.destination.oss_bucket_destination.role_arn},'
          f' bucket: {result.inventory_configuration.destination.oss_bucket_destination.bucket},'
          f' prefix: {result.inventory_configuration.destination.oss_bucket_destination.prefix},'
          f' lower size bound: {result.inventory_configuration.filter.lower_size_bound},'
          f' upper size bound: {result.inventory_configuration.filter.upper_size_bound},'
          f' storage class: {result.inventory_configuration.filter.storage_class},'
          f' prefix: {result.inventory_configuration.filter.prefix},'
          f' last modify begin time stamp: {result.inventory_configuration.filter.last_modify_begin_time_stamp},'
          f' last modify end time stamp: {result.inventory_configuration.filter.last_modify_end_time_stamp},'
    )

if __name__ == "__main__":
    main()

List inventory configurations

client.list_bucket_inventory(oss.ListBucketInventoryRequest(...))

A single request returns up to 100 configurations. If the response has is_truncated = True, pass next_continuation_token as the continuation_token in the next request to retrieve the remaining configurations.

Required parameters

Optional parameters

Response fields

result.status_code
result.request_id
result.list_inventory_configurations_result.is_truncated
result.list_inventory_configurations_result.next_continuation_token
result.list_inventory_configurations_result.inventory_configurations[]
  .id
  .is_enabled
  .included_object_versions
  .optional_fields
  .destination
  .schedule
  .filter

Example

The following example retrieves all inventory configurations by looping through pages until is_truncated is False.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="list bucket inventory sample")
parser.add_argument('--region', required=True)
parser.add_argument('--bucket', required=True)
parser.add_argument('--endpoint')

def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    continuation_token = None

    while True:
        result = client.list_bucket_inventory(oss.ListBucketInventoryRequest(
            bucket=args.bucket,
            continuation_token=continuation_token,
        ))

        print(f'status code: {result.status_code},'
              f' request id: {result.request_id},'
              f' is truncated: {result.list_inventory_configurations_result.is_truncated},'
              f' next continuation token: {result.list_inventory_configurations_result.next_continuation_token},'
        )

        if result.list_inventory_configurations_result.inventory_configurations:
            for r in result.list_inventory_configurations_result.inventory_configurations:
                print(f'id: {r.id},'
                      f' is enabled: {r.is_enabled},'
                      f' included object versions: {r.included_object_versions},'
                      f' optional fields: {r.optional_fields},'
                      f' destination: {r.destination},'
                      f' schedule: {r.schedule},'
                      f' filter: {r.filter}'
                )

        if not result.list_inventory_configurations_result.is_truncated:
            break

        continuation_token = result.list_inventory_configurations_result.next_continuation_token

if __name__ == "__main__":
    main()

Delete an inventory configuration

client.delete_bucket_inventory(oss.DeleteBucketInventoryRequest(...))

Required parameters

Example

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="delete bucket inventory sample")
parser.add_argument('--region', required=True)
parser.add_argument('--bucket', required=True)
parser.add_argument('--endpoint')
parser.add_argument('--inventory_id', required=True)

def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    result = client.delete_bucket_inventory(oss.DeleteBucketInventoryRequest(
        bucket=args.bucket,
        inventory_id=args.inventory_id,
    ))

    print(f'status code: {result.status_code}, request id: {result.request_id}')

if __name__ == "__main__":
    main()