Best Python code snippet using localstack_python
SG2PL-Batchsync.py
Source:SG2PL-Batchsync.py
...195 # Get the Prefix List info, specifically the current list version196 version = pl_info(pl, ec2remote)['Version']197 # Attempt to call the modify_managed_prefix_list with the current page of entries to remove and log the return198 try:199 response = ec2remote.modify_managed_prefix_list(200 PrefixListId=pl,201 CurrentVersion=version,202 RemoveEntries=page203 )204 version = response['PrefixList']['Version']205 message = "modify_managed_prefix_list call result: " + str(response)206 log_handler(message, 1, False, False)207 # If an exception occurs, log it, send a message via SNS and terminate the function unsuccessfully.208 except Exception as e:209 message = "Removing new CIDRs from " + str(pl) + " was unsuccessful. Error returned: " + str(e)210 log_handler(message, 3, True, True)211 return version212# Add the CIDRs specified to the Prefix List specified213def add_cidr_to_pl(pl: str, ips: set, version: str, ec2remote):214 # Create an empty list to populate the IPs as CIDRs215 entriestoadd = list()216 # Loop through the IPs and append the list with a dict with the IP in CIDR notation as a /32217 for ip in ips:218 value = {'Cidr': ip + "/32"}219 entriestoadd.append(value)220 # Log the list of dicts created above221 message = "Trying to add: " + str(entriestoadd) + " to " + str(pl) + " using version " + str(version)222 log_handler(message, 1, False, False)223 # Paginate the list into separate lists no greater than 99 CIDRs each to account for the maximum of 100224 # entries in AddEntries per modify_managed_prefix_list call225 paginatedentriestoadd = [entriestoadd[i:i + 99] for i in range(0, len(entriestoadd), 99)]226 # Log the paginated list of lists created above227 message = "Paginated list of list:: trying to remove: " + str(paginatedentriestoadd) + " from " \228 + str(pl) + " using version " + str(version)229 log_handler(message, 1, False, False)230 # Loop through the paginated list of lists for entries to add231 for page in paginatedentriestoadd:232 # Determine the current status of the Prefix List and if it is not in an acceptable state enter a loop to233 # check every second until it is ready.234 plstatus = pl_ready(pl, ec2remote)235 while not plstatus:236 time.sleep(1)237 plstatus = pl_ready(pl, ec2remote)238 # Get the Prefix List info, specifically the current list version239 version = pl_info(pl, ec2remote)['Version']240 # Attempt to call the modify_managed_prefix_list with the current page of entries to add and log the return241 try:242 response = ec2remote.modify_managed_prefix_list(243 PrefixListId=pl,244 CurrentVersion=version,245 AddEntries=page246 )247 version = response['PrefixList']['Version']248 message = "modify_managed_prefix_list call result: "+str(response)249 log_handler(message, 1, False, False)250 # If an exception occurs, log it, send a message via SNS and terminate the function unsuccessfully.251 except Exception as e:252 message = "Adding new CIDRs to "+str(pl)+" was unsuccessful. Error returned: "+str(e)253 log_handler(message, 3, True, True)254 return version255# Update CIDRs in the PL256def update_cidrs_in_pl(pl: str, ipstoremove: set, ipstoadd: set, version: str, ec2remote):257 # Create an empty list to populate the IPs as CIDRs to add258 entriestoadd = list()259 # Loop through the IPs to add and append the list with a dict with the IP in CIDR notation as a /32260 for iptoadd in ipstoadd:261 value = {'Cidr': iptoadd + "/32"}262 entriestoadd.append(value)263 # Log the list of dicts created above264 message = "Bulk update: Trying to add: " + str(entriestoadd) + " to " + str(pl) + " using version " + str(version)265 log_handler(message, 1, False, False)266 # Create an empty list to populate the IPs as CIDRs to remove267 entriestoremove = list()268 # Loop through the IPs to remove and append the list with a dict with the IP in CIDR notation as a /32269 for iptoremove in ipstoremove:270 value = {'Cidr': iptoremove + "/32"}271 entriestoremove.append(value)272 # Log the list of dicts created above273 message = "Bulk update: Trying to remove: "+str(entriestoremove)+" from "+str(pl)+" using version "+str(version)274 log_handler(message, 1, False, False)275 # Attempt to call the modify_managed_prefix_list with the entries to add and remove and log the return276 try:277 response = ec2remote.modify_managed_prefix_list(278 PrefixListId=pl,279 CurrentVersion=version,280 AddEntries=entriestoadd,281 RemoveEntries=entriestoremove282 )283 message = "modify_managed_prefix_list call result: " + str(response)284 log_handler(message, 1, False, False)285 return response['PrefixList']['Version']286 # If an exception occurs, log it, send a message via SNS and terminate the function unsuccessfully.287 except Exception as e:288 message = "Updating new CIDRs in " + str(pl) + " was unsuccessful. Error returned: " + str(e)289 log_handler(message, 3, True, True)290# Get the status of the Prefix List291def pl_ready(pl: str, ec2remote):292 # Call describe_managed_prefix_lists293 plinfodata = ec2remote.describe_managed_prefix_lists(294 PrefixListIds=[295 pl,296 ]297 )298 # Log the return299 message = "modify_managed_prefix_list call result: "+str(plinfodata)300 log_handler(message, 1, False, False)301 # Parse the current State302 plstate = plinfodata['PrefixLists'][0]['State']303 # Go through various known and documented state possibilities and return True if the Prefix List is in a state304 # ready for modification, False if it is in a state not ready for modification but should naturally turn to a305 # state that is ready later, or call the error handler with a message to send via SNS and terminate the Function306 # if it is in a state that requires user intervention before proceeding.307 if plstate == 'create-in-progress':308 message = str(pl)+" PL is in a creating state."309 log_handler(message, 3, False, True)310 elif plstate == 'create-complete':311 return True312 elif plstate == 'create-failed':313 message = "The prefix list "+str(pl)+" is in a state of creation failed. Please delete remove the" \314 "configuration for this sync and recreate it with a new prefix list."315 log_handler(message, 3, True, True)316 elif plstate == 'modify-in-progress':317 return False318 elif plstate == 'modify-complete':319 return True320 elif plstate == 'modify-failed':321 message = "There was an error modifying prefix list "+str(pl)+". The reason was: "\322 + str(plinfodata['PrefixLists'][0]['StateMessage'])323 log_handler(message, 3, True, True)324 elif plstate == 'restore-in-progress':325 return False326 elif plstate == 'restore-complete':327 return True328 elif plstate == 'restore-failed':329 message = "The prefix list "+str(pl)+" is in a state of restore-failed. Please correct this so the "\330 "status becomes modify-complete for the sync to resume successfully"331 log_handler(message, 3, True, True)332 elif plstate == 'delete-in-progress':333 message = "The prefix list "+str(pl)+" is currently being deleted. Please delete the AutoSG2PL "\334 "configuration to prevent unnecessary invocations of the associated Lambda functions."335 log_handler(message, 3, True, True)336 elif plstate == 'delete-complete':337 message = "The prefix list "+str(pl)+" has been deleted. Please delete the AutoSG2PL configuration "\338 "to prevent unnecessary invocations of the associated Lambda functions."339 log_handler(message, 3, True, True)340 elif plstate == 'delete-failed':341 message = "The prefix list "+str(pl)+" has been unsuccessfully deleted. Please either do an update "\342 "on the prefix list or delete it completely and delete the AutoSG2PL configuration to "\343 "prevent unnecessary invocations of the associated Lambda functions."344 log_handler(message, 3, True, True)345 else:346 message = "The prefix list "+str(pl)+" is in an unknown state of: "+str(plstate)+" Please make an "\347 "update so it gets into a modify-complete state."348 log_handler(message, 3, True, True)349# Resize the Prefix List to account for Max Entries350def pl_resize(pl: str, currentquota: int, newplsize: int, ec2remote):351 # Determine the optimal new size based on the configured padding percentage352 newmaxent = math.ceil(newplsize / sgrulemaxutilizationpercentage)353 # If the optimal new size exceeds the quota, send a warning via SNS and resize to 1 less than the current quota354 if newmaxent > currentquota:355 message = "The prefix list " + str(pl) + " was resized to accommodate the latest sync but below the padding " \356 "percentage threshold set due to the current Max Entries per Security Group quota. Please request an " \357 "increase in Service Quotas to allow for greater than " + str(newmaxent) + " entries per Security Group " \358 "or lower your security_group_quota_padding_percentage threshold configured on the AutoSG2PL Batch Sync " \359 "Lambda Function under Environment Variables"360 log_handler(message, 2, True, False)361 newmaxent = int(currentquota - 1)362 # Try to resize the Prefix List to the new value. If the new value exceeds a quota in another account the error363 # is handled by the Prefix List resize function itself and then caught during the Prefix List status check.364 try:365 response = ec2remote.modify_managed_prefix_list(366 PrefixListId=pl,367 MaxEntries=newmaxent368 )369 message = "modify_managed_prefix_list max entries call result: " + str(response)370 log_handler(message, 1, False, False)371 return response['PrefixList']['Version']372 # If an exception occurs, log it, send a message via SNS and terminate the function unsuccessfully.373 except Exception as e:374 message = "Updating Max Entries for " + str(pl) + " was unsuccessful. Error returned: " + str(e)375 log_handler(message, 3, True, True)376# the main lambda function handler377def lambda_handler(event, context):378 # Get the supplied Security Group and store it in a variable. If none is supplied379 if 'sg' in event.keys():...
app.py
Source:app.py
...65 print("already in list so no action")66 else:67 print("add")68 if len(current_entries) + 1 != prefix_list["MaxEntries"]:69 response = client.modify_managed_prefix_list(70 PrefixListId=prefix_list_id,71 MaxEntries=len(current_entries) + 172 )73 sleep(3)74 response = client.modify_managed_prefix_list(75 PrefixListId=prefix_list_id,76 CurrentVersion=current_prefix_list_version,77 AddEntries=[78 {79 'Cidr': private_id_address + "/32",80 'Description': 'added by EventBridge Lambda'81 },82 ]83 )84 # if the instance state change is 'stopping' so we remove the private IP CIDR to the Prefix List85 elif ec2_state == "stopping":86 if is_in_list:87 print("remove")88 response = client.modify_managed_prefix_list(89 PrefixListId=prefix_list_id,90 CurrentVersion=current_prefix_list_version,91 RemoveEntries=[92 {93 'Cidr': private_id_address + "/32"94 },95 ]96 )97 if len(current_entries) != 1: 98 sleep(3)99 response = client.modify_managed_prefix_list(100 PrefixListId=prefix_list_id,101 MaxEntries=len(current_entries) - 1102 )103 else:104 print("not in list so no action")...
update-ipranges.py
Source:update-ipranges.py
...44 entry_list.append({'Cidr': ip})45 print("MODIFYING")46 print("ADDING", entry_list)47 print("REMOVING", to_remove)48 ec2.modify_managed_prefix_list(49 DryRun=False,50 PrefixListId=prefix_list_id,51 AddEntries=entry_list,52 RemoveEntries=to_remove,53 CurrentVersion=prefix_list_version54 )55 return {56 'statusCode': 200,57 'body': "Updated prefixlist"58 }59 else:60 print("Nothing to modify")61 return {62 'statusCode': 200,...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!