KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use delete_destination method in localstack

Best Python code snippet using localstack_python

s3CopySyncScript.py

Source:s3CopySyncScript.py

1import ast2import boto33import subprocess4import sys5import csv6from io import StringIO7from urllib.parse import urlparse8import logging9logging.basicConfig(10    level=logging.INFO,11    format='%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s',12    datefmt='%Y-%m-%d %H:%M:%S',13)14LOGGER = logging.getLogger()15def get_bool_input(input_parameter):16	"""Returns a boolean corresponding to a string "True/False" input17	18	Args:19	    input_parameter (str): String input for a boolean20	21	Returns:22	    bool: True/False input23	"""24	if input_parameter.lower().capitalize() in ["True","False"]:25		return ast.literal_eval(input_parameter.lower().capitalize())26	else:27		return None28class S3SyncUtils:29	def __init__(self):30		self.s3_resource = boto3.resource('s3')31	def __copy_if_exists(self,src_bucket, src_key, src_path,dst_path,delete_destination):32		s3_head_object_response = self.s3_resource.meta.client.head_object(33			Bucket=src_bucket,34			Key=src_key)35		if s3_head_object_response['ContentType'].startswith('application/x-directory'): # sync36			LOGGER.info('Source is a directory! Syncing entire paths.')37			self.__s3_sync_helper(src_path,dst_path,delete_destination)38		else:39			LOGGER.info('File Exists! Overwriting file')40			copy_source = {41				'Bucket': src_bucket,42				'Key': src_key43			}44			dst_bucket_key = dst_path.split('//')[1].split('/',1)45			self.s3_resource.meta.client.copy(CopySource=copy_source, 46						Bucket=dst_bucket_key[0], 47						Key=dst_bucket_key[1],48						ExtraArgs={49							'ACL':'bucket-owner-full-control'50						})51	def __s3_sync_helper(self, src_path, dst_path, delete_destination):52		sync_cmd = ['/usr/local/bin/aws',53				's3',54				'sync',55				src_path,56				dst_path,57				'--acl','bucket-owner-full-control']58		if delete_destination:59			sync_cmd.append('--delete')60		out = subprocess.run(sync_cmd, capture_output=True)61		out = out.stdout.decode()62		LOGGER.info(out)63	def s3_copy_sync(self,input_s3_bucket,input_s3_key, skip_header=True, delete_destination=False):64		"""65		:param input_s3_bucket: bucket name of the input file66		:param input_s3_key: name of the file67		:param skip_header:  boolean  : skip first row if true68		:param delete_destination: boolean : delete destination if true69		:return: None70		"""71		csv_str = self.s3_resource.Object(input_s3_bucket,input_s3_key).get()['Body'].read().decode('utf-8')72		rdr = csv.reader(StringIO(csv_str), delimiter=',')73		if skip_header:74			next(rdr) #skip the header row75		#loop over each file/path given as src/destination.76		for row in rdr:77			source=row[0]78			dest=row[1]79			LOGGER.info(f'Source:: {source} Dest:: {dest}')80			# split the S3 path for source into what would be bucket and key81			# EG: s3://bucket-name/some/path/file.txt --> bucket-name, some/path/file.txt82			parsed_s3_url = urlparse(source)83			bucket_src = parsed_s3_url.netloc84			key_src = parsed_s3_url.path[1:] # remove first "/" for a valid s3 key85			self.__copy_if_exists(bucket_src,key_src,source,dest,delete_destination)86def main(argv):87	if len(argv) != 5:88		LOGGER.info("Syntax: python s3CopySyncScript.py <<bucket containing input .csv file>> <<key of input .csv file>> <<header True/False>> <<sync_delete True/False>>")89		sys.exit(1)90	LOGGER.info(f"Received {len(argv)} arguments:\n"91		f"\tInput csv file bucket name ::: {argv[1]}\n"92		f"\tInput csv file file name ::: {argv[2]}\n"93		f"\tFile containing header ::: {argv[3]}\n"94		f"\tDelete destination files ::: {argv[4]}" )95	S3_UTILS = S3SyncUtils()96	header_input = get_bool_input(argv[3])97	delete_input = get_bool_input(argv[4])98	if header_input is None or delete_input is None:99		LOGGER.error("Incorrect input, the HEADER must be true/false and delete files must be true or false")100		sys.exit(2)101	S3_UTILS.s3_copy_sync(argv[1],argv[2],header_input,delete_input)102if __name__ == '__main__':...

urls.py

Source:urls.py

1"""csv_proj URL Configuration2The `urlpatterns` list routes URLs to views. For more information please see:3    https://docs.djangoproject.com/en/4.0/topics/http/urls/4Examples:5Function views6    1. Add an import:  from my_app import views7    2. Add a URL to urlpatterns:  path('', views.home, name='home')8Class-based views9    1. Add an import:  from other_app.views import Home10    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')11Including another URLconf12    1. Import the include() function: from django.urls import include, path13    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))14"""15from django.contrib import admin16from django.urls import path , include17from python_bot import views18from python_bot.views import Source_file , Compare_file19urlpatterns = [20    path('',views.index,name='index'),21    path('u/',Source_file.as_view()),22    path('d/',Compare_file.as_view()),23    path('compare/',views.compare,name='compare'),24    path('destination_list/',views.show_all_destination_files,name='show_all_destination_files'),25    path('source_list/',views.show_all_source_files,name='show_all_source_files'),26    path('delete_source/',views.delete_source,name='delete_source'),27    path('delete_destination/',views.delete_destination,name='delete_destination'),28    path('try/',views.try_func,name='try_func')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.