Answered step by step
Verified Expert Solution
Link Copied!

Question

1 Approved Answer

Hello. I am trying to finish up a Python program in AWS that access S3 to make and change items in different buckets. When I

Hello. I am trying to finish up a Python program in AWS that access S3 to make and change items in different buckets. When I try to move a file from one bucket to another (menu option 4), once I've chosen my buckets and file, I get the following error:

botocore.exceptions.ClientError: An error occurred (404) when calling the HeadObject operation: Not Found

I believe there's an issue with my move_object function. I've tried adjusting the code to access S3 based on other functions, but I keep getting errors.

There are other code errors, but fixing this issue will get the program generally functioning the way I need it to. Any help with these things is greatly appreciated.

from datetime import datetime import copy import logging import boto3 from boto3 import client import sys import botocore from botocore.exceptions import ClientError

def create_bucket(bucket_name, region=None): # creates new bucket try: if region is None: s3_client = boto3.client('s3') s3_client.create_bucket(Bucket=bucket_name) else: s3_client = boto3.client('s3', region_name=region) location = {'LocationConstraint': region} s3_client.create_bucket(Bucket=bucket_name,CreateBucketConfiguration=location) except ClientError as e: logging.error(e) return False return True

def bucket_exists(bucket_name): # looks to see if bucket exists s3 = boto3.client('s3') try: response = s3.head_bucket(Bucket=bucket_name) except ClientError as e: logging.debug(e) return False return True

def list_bucket_objects(bucket_name): # list of objects in chosen bucket s3 = boto3.client('s3') try: response = s3.list_objects_v2(Bucket=bucket_name) except ClientError as e: # AllAccessDisabled error == bucket not found print("Bucket does not exists. Please try again.") # if object dows not exist, inform user menu() return response['Contents']

def get_object(bucket_name, object_name): #get the object

# Retrieve the object s3 = boto3.client('s3') try: response = s3.get_object(Bucket=bucket_name, Key=object_name) except ClientError as e: print("Object doesn't exist. Please try again.") #inform user object does not exist menu() # Return an open StreamingBody object return response['Body']

def delete_object(bucket_name, object_name): # Delete an object from an S3 bucket

# Delete the object s3 = boto3.client('s3') try: s3.delete_object(Bucket=bucket_name, Key=object_name) except ClientError as e: logging.error(e) return False return True

def move_object(bucket_name, bucket_name2, object_name): #moves obj from one bucket to another s3 = boto3.resource('s3') copy_source = { 'Bucket': bucket_name, 'Key': object_name } s3.meta.client.copy(copy_source, bucket_name2, object_name)

def download(bucket_name, object_name): #downloads object from bucket BUCKET_NAME = bucket_name KEY = object_name

s3 = boto3.resource('s3')

try: s3.Bucket(BUCKET_NAME).download_file(KEY, 'my_local_image.jpg') except botocore.exceptions.ClientError as e: if e.response['Error']['Code'] == "404": print("The object does not exist.") else: raise

# Defining variables for datetime now=datetime.now() print("now=", now) dt_string = now.strftime("%d/%m/%Y %H:%M:%S")

def menu(): print(" Welcome to the AWS S3 Bucket Applicaiton ") print("Please choose from the following: ") print("1. Create new bucket") print("2. Add object in a bucket") print("3. Delete object in a bucket") print("4. Move object into different bucket") print("5. Download object from bucket") print("6. Exit Program ")

choice = str(input())

if choice == '1': #user enter name of bucket print("Enter the name of your bucket: ") bucket_name = str(input()) # Check if the bucket exists if bucket_exists(bucket_name): logging.info(f'{bucket_name} already exists. Please try again.') menu() else: create_bucket(bucket_name) print(" ", bucket_name, " has been created.") menu()

elif choice == '2': # Create an S3 client s3 = boto3.client('s3') #lists available buckets response = s3.list_buckets() buckets = [bucket['Name'] for bucket in response['Buckets']] print("Buckets: %s " % buckets) print("Type a bucket you would like to use: ") bucket_name = str(input()) if bucket_exists(bucket_name): #if bucket exists, choose file to upload print("Choose what file you want to upload") file_name = str(input()) else: print("Bucket does not exist. Please try again.") menu() s3.upload_file(file_name, bucket_name, file_name) print(file_name, " has been uploaded to ", bucket_name) menu()

elif choice == '3': # Create an S3 client s3 = boto3.client('s3') # Call S3 to list current buckets response = s3.list_buckets() # Get a list of all bucket names from the response buckets = [bucket['Name'] for bucket in response['Buckets']]

# Print out the bucket list print("Bucket List: %s" % buckets) print(" Type a bucket you would like to use: ") bucket_name = str(input()) if bucket_exists(bucket_name): #if bucket exists print(list_bucket_objects(bucket_name)) #print obj list print(" Type the object you want deleted: ") object_name = str(input()) if get_object(bucket_name, object_name): #if object exists delete_object(bucket_name, object_name) #delete object print(object_name, " has been deleted.") menu() else: print("Bucket does not exist. Please try again.") menu()

elif choice == '4': # Create an S3 client s3 = boto3.client('s3') # Call S3 to list current buckets response = s3.list_buckets() # Get a list of all bucket names from the response buckets = [bucket['Name'] for bucket in response['Buckets']]

# Print out the bucket list print("Bucket List: %s" % buckets) print("Type a bucket you would like to use: ") bucket_name = str(input()) print(list_bucket_objects(bucket_name)) print("Type the object you want moved: ") object_name = str(input()) print("Bucket List: %s" % buckets) print("Type a bucket you would want the object to move to: ") bucket_name2= str(input()) move_object(bucket_name, bucket_name2, object_name) print(object_name, " has been moved to ", bucket_name2) elif choice == '5': # Create an S3 client s3 = boto3.client('s3') # Call S3 to list current buckets response = s3.list_buckets() # Get a list of all bucket names from the response buckets = [bucket['Name'] for bucket in response['Buckets']]

# Print out the bucket list print("Bucket List: %s" % buckets) print("Type a bucket you would like to use: ") bucket_name = str(input()) print(list_bucket_objects(bucket_name)) print("Choose what file you want to download: ") object_name = str(input()) download(bucket_name, object_name) print(object_name, " is downloading now.") menu() elif choice == '6': # Ends application print("Thank you for using the AWS S3 Bucket Applicaiton") print("Date and Time =", dt_string) sys.exit(0) # ends program else: # alerts user of invalid entry print("Invalid entry. Please try again.") menu() if __name__ == '__main__': menu()

Step by Step Solution

There are 3 Steps involved in it

Step: 1

blur-text-image

Get Instant Access to Expert-Tailored Solutions

See step-by-step solutions with expert insights and AI powered tools for academic success

Step: 2

blur-text-image

Step: 3

blur-text-image

Ace Your Homework with AI

Get the answers you need in no time with our AI-driven, step-by-step assistance

Get Started

Recommended Textbook for

Relational Database Technology

Authors: Suad Alagic

1st Edition

354096276X, 978-3540962762

More Books

Students also viewed these Databases questions

Question

3. What are potential solutions?

Answered: 1 week ago