1. import boto3
2. from os import getenv
3. from dotenv import load_dotenv
4. import logging
5. from botocore.exceptions import ClientError
6. from hashlib import md5
7. from time import localtime
8.  
9. load_dotenv()
10.  
11. def init_client():
12.     try:
13.         client = boto3.client("s3",
14.                               aws_access_key_id=getenv("aws_access_key_id"),
15.                               aws_secret_access_key=getenv(
16.                                   "aws_secret_access_key"),
17.                               aws_session_token=getenv("aws_session_token"),
18.                               region_name=getenv("aws_region_name")
19.                               #  config=botocore.client.Config(
20.                               #      connect_timeout=conf.remote_cfg["remote_timeout"],
21.                               #      read_timeout=conf.remote_cfg["remote_timeout"],
22.                               #      region_name=conf.remote_cfg["aws_default_region"],
23.                               #      retries={
24.                               #          "max_attempts": conf.remote_cfg["remote_retries"]}
25.                               )
26.         # check if credentials are correct
27.         client.list_buckets()
28.  
29.         return client
30.     except ClientError as e:
31.         logging.error(e)
32.     except:
33.         logging.error("Unexpected error")
34.  
35.  
36. def list_buckets(aws_s3_client):
37.     try:
38.         # https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListBuckets.html
39.         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/list_buckets.html
40.         return aws_s3_client.list_buckets()
41.     except ClientError as e:
42.         logging.error(e)
43.         return False
44.  
45.  
46. def create_bucket(aws_s3_client, bucket_name, region=getenv("aws_region_name")):
47.     # Create bucket
48.     try:
49.         location = {'LocationConstraint': region}
50.         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/create_bucket.html
51.         response = aws_s3_client.create_bucket(
52.             Bucket=bucket_name,
53.             CreateBucketConfiguration=location
54.         )
55.     except ClientError as e:
56.         logging.error(e)
57.         return False
58.     status_code = response["ResponseMetadata"]["HTTPStatusCode"]
59.     if status_code == 200:
60.         return True
61.     return False
62.  
63.  
64. def delete_bucket(aws_s3_client, bucket_name):
65.     # Delete bucket
66.     try:
67.         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/delete_bucket.html
68.         response = aws_s3_client.delete_bucket(Bucket=bucket_name)
69.     except ClientError as e:
70.         logging.error(e)
71.         return False
72.     status_code = response["ResponseMetadata"]["HTTPStatusCode"]
73.     if status_code == 200:
74.         return True
75.     return False
76.  
77.  
78. def bucket_exists(aws_s3_client, bucket_name):
79.     try:
80.         response = aws_s3_client.head_bucket(Bucket=bucket_name)
81.     except ClientError as e:
82.         logging.error(e)
83.         return False
84.     status_code = response["ResponseMetadata"]["HTTPStatusCode"]
85.     if status_code == 200:
86.         return True
87.     return False
88.  
89.  
90. def download_file_and_upload_to_s3(aws_s3_client, bucket_name, url, file_name, keep_local=False):
91.     from urllib.request import urlopen
92.     import io
93.     with urlopen(url) as response:
94.         content = response.read()
95.         try:
96.             aws_s3_client.upload_fileobj(
97.                 Fileobj=io.BytesIO(content),
98.                 Bucket=bucket_name,
99.                 ExtraArgs={'ContentType': 'image/jpg'},
100.                 Key=file_name
101.             )
102.         except Exception as e:
103.             logging.error(e)
104.  
105.     if keep_local:
106.         with open(file_name, mode='wb') as jpg_file:
107.             jpg_file.write(content)
108.  
109.     # public URL
110.     return "https://s3-{0}.amazonaws.com/{1}/{2}".format(
111.         'us-west-2',
112.         bucket_name,
113.         file_name
114.     )
115.  
116. def set_object_access_policy(aws_s3_client, bucket_name, file_name):
117.     try:
118.         response = aws_s3_client.put_object_acl(
119.             ACL="public-read",
120.             Bucket=bucket_name,
121.             Key=file_name
122.         )
123.     except ClientError as e:
124.         logging.error(e)
125.         return False
126.     status_code = response["ResponseMetadata"]["HTTPStatusCode"]
127.     if status_code == 200:
128.         return True
129.     return False
130.  
131. def generate_public_read_policy(bucket_name):
132.     import json
133.     policy = {
134.         "Version": "2012-10-17",
135.         "Statement": [
136.             {
137.                 "Sid": "PublicReadGetObject",
138.                 "Effect": "Allow",
139.                 "Principal": "*",
140.                 "Action": "s3:GetObject",
141.                 "Resource": f"arn:aws:s3:::{bucket_name}/*",
142.             }
143.         ],
144.     }
145.  
146.     return json.dumps(policy)
147.  
148. def create_bucket_policy(aws_s3_client, bucket_name):
149.     aws_s3_client.put_bucket_policy(
150.         Bucket=bucket_name, Policy=generate_public_read_policy(bucket_name)
151.     )
152.     print("Bucket policy created successfully")
153.  
154. def read_bucket_policy(aws_s3_client, bucket_name):
155.     try:
156.         policy = aws_s3_client.get_bucket_policy(Bucket=bucket_name)
157.         policy_str = policy["Policy"]
158.         print(policy_str)
159.     except ClientError as e:
160.         logging.error(e)
161.         return False
162.  
163. if __name__ == "__main__":
164.     s3_client = init_client()
165.  
166.     # print(download_file_and_upload_to_s3(
167.     #     s3_client, 'new-bucket-btu',
168.     #     'https://www.coreldraw.com/static/cdgs/images/free-trials/img-ui-cdgsx.jpg',
169.     #     f'image_file_{md5(str(localtime()).encode("utf-8")).hexdigest()}.jpg',
170.     #     keep_local=True
171.     # ))
172.     # create_bucket_policy(s3_client, 'new-bucket-btu')
173.     # read_bucket_policy(s3_client, 'new-bucket-btu')
174.  
175.     # buckets = list_buckets(s3_client)
176.     # if buckets:
177.     #     for bucket in buckets['Buckets']:
178.     #         print(f'  {bucket["Name"]}')
179.  
180.  
181.     # print(f'created bucket status: { create_bucket(s3_client, "new-bucket-1-btu")}')
182.     # print(f'deleted bucket status: { delete_bucket(s3_client, "btudevopsteam1")}')
183.     # print(f'Bucket exists: { bucket_exists(s3_client, "automatinawsbttu-commandline")}')
184.  
185.     # print(f"set read status: {set_object_access_policy(s3_client, 'new-bucket-btu', 'image_file_78bc222b20d1ff69cdf1290a7537d5fd.jpg')}")
186.  
187.