Best Python code snippet using localstack_python
tasks.py
Source:tasks.py
1import requests2# import logging3from rq import get_current_job4from django_rq import job5from django.core.management import call_command6from shared_foundation.models import SharedOrganization7from shared_foundation.utils import get_point_from_ip8# logger = logging.getLogger(__name__)9@job10def process_associate_with_slug_func(schema_name, slug):11 call_command('process_associate_with_slug', schema_name, slug, verbosity=0)12@job13def geocode_associate_address_func(schema_name, associate_slug):14 call_command('geocode_associate_address', schema_name, associate_slug, verbosity=0)15@job16def geoip2_associate_audit_func(organization, associate):17 organization.activate_tenant()18 if associate.created_from:19 associate.created_from_position = get_point_from_ip(associate.created_from)20 if associate.last_modified_from:21 associate.last_modified_from_position = get_point_from_ip(associate.last_modified_from)22 associate.save()23@job24def geoip2_associate_address_audit_func(organization, associate_address):25 organization.activate_tenant()26 if associate_address.created_from:27 associate_address.created_from_position = get_point_from_ip(associate_address.created_from)28 if associate_address.last_modified_from:29 associate_address.last_modified_from_position = get_point_from_ip(associate_address.last_modified_from)30 associate_address.save()31@job32def geoip2_associate_contact_audit_func(organization, associate_contact):33 organization.activate_tenant()34 if associate_contact.created_from:35 associate_contact.created_from_position = get_point_from_ip(associate_contact.created_from)36 if associate_contact.last_modified_from:37 associate_contact.last_modified_from_position = get_point_from_ip(associate_contact.last_modified_from)38 associate_contact.save()39@job40def geoip2_associate_metric_audit_func(organization, associate_metric):41 organization.activate_tenant()42 if associate_metric.created_from:43 associate_metric.created_from_position = get_point_from_ip(associate_metric.created_from)44 if associate_metric.last_modified_from:45 associate_metric.last_modified_from_position = get_point_from_ip(associate_metric.last_modified_from)46 associate_metric.save()47@job48def geoip2_associate_comment_audit_func(organization, associate_comment):49 organization.activate_tenant()50 if associate_comment.comment:51 comment = associate_comment.comment52 if comment.created_from:53 comment.created_from_position = get_point_from_ip(comment.created_from)54 if comment.last_modified_from:55 comment.last_modified_from_position = get_point_from_ip(comment.last_modified_from)...
callback_aws.py
Source:callback_aws.py
...20 return wrapped21def get_instance_metadata():22 return boto.utils.get_instance_identity()['document']23@retry24def associate_address(ec2, allocation_id, instance_id):25 return ec2.associate_address(instance_id=instance_id, allocation_id=allocation_id, allow_reassociation=True)26@retry27def tag_resource(ec2, resource_id, tags):28 return ec2.create_tags([resource_id], tags)29@retry30def list_volumes(ec2, instance_id):31 return ec2.get_all_volumes(filters={'attachment.instance-id': instance_id})32@retry33def get_instance(ec2, instance_id):34 return ec2.get_only_instances([instance_id])[0]35def main():36 logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)37 # EIP_ALLOCATION is optional argument38 argc = len(sys.argv)39 if argc not in (4, 5) or sys.argv[argc - 3] not in ('on_start', 'on_stop', 'on_role_change'):40 sys.exit("Usage: {0} [eip_allocation_id] action role name".format(sys.argv[0]))41 action, role, cluster = sys.argv[argc - 3:argc]42 metadata = get_instance_metadata()43 instance_id = metadata['instanceId']44 ec2 = boto.ec2.connect_to_region(metadata['region'])45 if argc == 5 and role in ('master', 'standby_leader') and action in ('on_start', 'on_role_change'):46 associate_address(ec2, sys.argv[1], instance_id)47 instance = get_instance(ec2, instance_id)48 tags = {'Role': role}49 tag_resource(ec2, instance_id, tags)50 tags.update({'Instance': instance_id})51 volumes = list_volumes(ec2, instance_id)52 for v in volumes:53 if 'Name' in v.tags:54 tags_to_update = tags55 else:56 if v.attach_data.device == instance.root_device_name:57 volume_device = 'root'58 else:59 volume_device = 'data'60 tags_to_update = dict(tags, Name='spilo_{}_{}'.format(cluster, volume_device))...
lesson4.py
Source:lesson4.py
...18 allocation = self.ec2.allocate_address(Domain='vpc')19 return allocation20 except ClientError as e:21 print(e)22 def associate_address(self, allocation_id, instance_id):23 try:24 response = self.ec2.associate_address(AllocationId=allocation_id,25 InstanceId=instance_id)26 # print(response)27 except ClientError as e:28 print(e)29 def release_address(self, allocation_id):30 try:31 self.ec2.release_address(AllocationId=allocation_id)32 print('Address released')33 except ClientError as e:34 print(e)35if __name__ == "__main__":36 aws_access_key_id = ""37 aws_secret_access_key = ""38 region_name = "us-east-1"39 eip = ElasticIP(aws_access_key_id,40 aws_secret_access_key,41 region_name)42 # print(eip.describe_address())43 # print(eip.get_allocation_id())44 # eip.associate_address(allocation_id=eip.get_allocation_id(),45 # instance_id=eip.get_instance_id())...
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!!