Dev sample

sample/aio/get_bucket_info.py

@@ -0,0 +1,63 @@
+import asyncio
+import argparse
+import alibabacloud_oss_v2 as oss
+import alibabacloud_oss_v2.aio as oss_aio
+
+
+parser = argparse.ArgumentParser(description="get bucket info sample")
+parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
+parser.add_argument('--bucket', help='The name of the bucket.', required=True)
+parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
+
+
+async def main():
+
+    args = parser.parse_args()
+
+    # Loading credentials values from the environment variables
+    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
+
+    # Using the SDK's default configuration
+    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_aio.AsyncClient(cfg)
+
+    result = await client.get_bucket_info(oss.GetBucketInfoRequest(
+        bucket=args.bucket,
+    ))
+
+    print(f'status code: {result.status_code},'
+          f' request id: {result.request_id},'
+          f' name: {result.bucket_info.name},' 
+          f' access monitor: {result.bucket_info.access_monitor},' 
+          f' location: {result.bucket_info.location},' 
+          f' creation date: {result.bucket_info.creation_date},' 
+          f' extranet endpoint: {result.bucket_info.extranet_endpoint},' 
+          f' intranet endpoint: {result.bucket_info.intranet_endpoint},' 
+          f' acl: {result.bucket_info.acl},' 
+          f' data redundancy type: {result.bucket_info.data_redundancy_type},' 
+          f' id: {result.bucket_info.owner.id},' 
+          f' display name: {result.bucket_info.owner.display_name},' 
+          f' storage class: {result.bucket_info.storage_class},' 
+          f' resource group id: {result.bucket_info.resource_group_id},' 
+          f' kms master key id: {result.bucket_info.sse_rule.kms_master_key_id},' 
+          f' sse algorithm: {result.bucket_info.sse_rule.sse_algorithm},' 
+          f' kms data encryption: {result.bucket_info.sse_rule.kms_data_encryption},' 
+          f' versioning: {result.bucket_info.versioning},' 
+          f' transfer acceleration: {result.bucket_info.transfer_acceleration},' 
+          f' cross region replication: {result.bucket_info.cross_region_replication},' 
+          f' log bucket: {result.bucket_info.bucket_policy.log_bucket},' 
+          f' log prefix: {result.bucket_info.bucket_policy.log_prefix},' 
+          f' comment: {result.bucket_info.comment},' 
+          f' block public access: {result.bucket_info.block_public_access},'
+    )
+
+    await client.close() 
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sample/aio/get_object.py

@@ -0,0 +1,68 @@
+import asyncio
+import argparse
+import alibabacloud_oss_v2 as oss
+import alibabacloud_oss_v2.aio as oss_aio
+
+parser = argparse.ArgumentParser(description="get object sample")
+parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
+parser.add_argument('--bucket', help='The name of the bucket.', required=True)
+parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
+parser.add_argument('--key', help='The name of the object.', required=True)
+
+
+async def main():
+
+    args = parser.parse_args()
+
+    # Loading credentials values from the environment variables
+    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
+
+    # Using the SDK's default configuration
+    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_aio.AsyncClient(cfg)
+
+    result = await client.get_object(oss.GetObjectRequest(
+        bucket=args.bucket,
+        key=args.key,
+    ))
+
+    # Under the async mode, the data has been loaded into memory
+    #print(f'content:{result.body.content}')
+
+    print(f'status code: {result.status_code},'
+          f' request id: {result.request_id},'
+          f' content length: {result.content_length},' 
+          f' content range: {result.content_range},' 
+          f' content type: {result.content_type},' 
+          f' etag: {result.etag},' 
+          f' last modified: {result.last_modified},' 
+          f' content md5: {result.content_md5},' 
+          f' cache control: {result.cache_control},' 
+          f' content disposition: {result.content_disposition},' 
+          f' content encoding: {result.content_encoding},' 
+          f' expires: {result.expires},' 
+          f' hash crc64: {result.hash_crc64},' 
+          f' storage class: {result.storage_class},' 
+          f' object type: {result.object_type},' 
+          f' version id: {result.version_id},' 
+          f' tagging count: {result.tagging_count},' 
+          f' server side encryption: {result.server_side_encryption},' 
+          f' server side data encryption: {result.server_side_data_encryption},' 
+          f' server side encryption key id: {result.server_side_encryption_key_id},' 
+          f' next append position: {result.next_append_position},' 
+          f' expiration: {result.expiration},' 
+          f' restore: {result.restore},' 
+          f' process status: {result.process_status},' 
+          f' delete marker: {result.delete_marker},'
+    )
+
+    await client.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sample/aio/put_object.py

@@ -0,0 +1,48 @@
+import asyncio
+import argparse
+import alibabacloud_oss_v2 as oss
+import alibabacloud_oss_v2.aio as oss_aio
+
+parser = argparse.ArgumentParser(description="put object sample")
+parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
+parser.add_argument('--bucket', help='The name of the bucket.', required=True)
+parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
+parser.add_argument('--key', help='The name of the object.', required=True)
+
+async def main():
+
+    args = parser.parse_args()
+
+    # Loading credentials values from the environment variables
+    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
+
+    # Using the SDK's default configuration
+    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_aio.AsyncClient(cfg)
+
+    data = b'hello world'
+
+    result = await client.put_object(oss.PutObjectRequest(
+        bucket=args.bucket,
+        key=args.key,
+        body=data,
+    ))
+
+    print(f'status code: {result.status_code},'
+          f' request id: {result.request_id},'
+          f' content md5: {result.content_md5},' 
+          f' etag: {result.etag},' 
+          f' hash crc64: {result.hash_crc64},' 
+          f' version id: {result.version_id},'
+    )
+
+    await client.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())