s3_client.py
Go to the documentation of this file.
1 #!/usr/bin/env python
2 # Copyright (c) 2020, Amazon.com, Inc. or its affiliates. All Rights Reserved.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License").
5 # You may not use this file except in compliance with the License.
6 # A copy of the License is located at
7 #
8 # http://aws.amazon.com/apache2.0
9 #
10 # or in the "license" file accompanying this file. This file is distributed
11 # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12 # express or implied. See the License for the specific language governing
13 # permissions and limitations under the License.
14 import boto3
15 
16 from botocore.client import Config
17 from botocore.exceptions import ClientError
18 
19 _SIGV4_NAME = 's3v4'
20 
21 class S3Client(object):
22  def __init__(self, region_name):
23  self.s3_client = boto3.client(
24  's3', region_name=region_name, config=Config(signature_version=_SIGV4_NAME))
25  self.s3_resource = boto3.resource('s3', region_name=region_name)
26  self.s3_region = region_name
27 
28  def create_bucket(self, bucket_name):
29  response = self.s3_client.create_bucket(
30  Bucket=bucket_name,
31  CreateBucketConfiguration={
32  'LocationConstraint': self.s3_region
33  }
34  )
35  return response['Location']
36 
37  def delete_bucket(self, bucket_name):
38  self.s3_client.delete_bucket(
39  Bucket=bucket_name
40  )
41 
42  def delete_all_objects(self, bucket_name):
43  bucket = self.s3_resource.Bucket(bucket_name)
44  bucket.objects.all().delete()
45 
46  def delete_objects(self, bucket_name, keys):
47  objects_to_delete = []
48  for key in keys:
49  objects_to_delete.append({
50  'Key': key.strip()
51  })
52  self.s3_client.delete_objects(
53  Bucket=bucket_name,
54  Delete={
55  'Objects': objects_to_delete
56  }
57  )
58 
59  def list_objects(self, bucket_name):
60  response = self.s3_client.list_objects(Bucket=bucket_name)
61  return response.get('Contents', [])
62 
63  def wait_for_bucket_create(self, bucket_name):
64  waiter = self.s3_client.get_waiter('bucket_exists')
65  waiter.wait(Bucket=bucket_name)
66 
67  def download_file(self, bucket_name, key, file_name):
68  self.s3_client.download_file(
69  Bucket=bucket_name,
70  Key=key,
71  Filename=file_name)
def create_bucket(self, bucket_name)
Definition: s3_client.py:28
def __init__(self, region_name)
Definition: s3_client.py:22
def delete_bucket(self, bucket_name)
Definition: s3_client.py:37
def delete_all_objects(self, bucket_name)
Definition: s3_client.py:42
def wait_for_bucket_create(self, bucket_name)
Definition: s3_client.py:63
def delete_objects(self, bucket_name, keys)
Definition: s3_client.py:46
def list_objects(self, bucket_name)
Definition: s3_client.py:59
def download_file(self, bucket_name, key, file_name)
Definition: s3_client.py:67


integ_tests
Author(s): AWS RoboMaker
autogenerated on Tue Feb 16 2021 03:52:56