<s> """<STR_LIT>""" <EOL> import base64 <EOL> import boto <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . regioninfo import RegionInfo , get_regions , load_regions <EOL> from boto . ec2 . autoscale . request import Request <EOL> from boto . ec2 . autoscale . launchconfig import LaunchConfiguration <EOL> from boto . ec2 . autoscale . group import AutoScalingGroup <EOL> from boto . ec2 . autoscale . group import ProcessType <EOL> from boto . ec2 . autoscale . activity import Activity <EOL> from boto . ec2 . autoscale . policy import AdjustmentType <EOL> from boto . ec2 . autoscale . policy import MetricCollectionTypes <EOL> from boto . ec2 . autoscale . policy import ScalingPolicy <EOL> from boto . ec2 . autoscale . policy import TerminationPolicies <EOL> from boto . ec2 . autoscale . instance import Instance <EOL> from boto . ec2 . autoscale . scheduled import ScheduledUpdateGroupAction <EOL> from boto . ec2 . autoscale . tag import Tag <EOL> from boto . ec2 . autoscale . limits import AccountLimits <EOL> from boto . compat import six <EOL> RegionData = load_regions ( ) . get ( '<STR_LIT>' , { } ) <EOL> def regions ( ) : <EOL> """<STR_LIT>""" <EOL> return get_regions ( '<STR_LIT>' , connection_cls = AutoScaleConnection ) <EOL> def connect_to_region ( region_name , ** kw_params ) : <EOL> """<STR_LIT>""" <EOL> for region in regions ( ) : <EOL> if region . name == region_name : <EOL> return region . connect ( ** kw_params ) <EOL> return None <EOL> class AutoScaleConnection ( AWSQueryConnection ) : <EOL> APIVersion = boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> DefaultRegionEndpoint = boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> DefaultRegionName = boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def __init__ ( self , aws_access_key_id = None , aws_secret_access_key = None , <EOL> is_secure = True , port = None , proxy = None , proxy_port = None , <EOL> proxy_user = None , proxy_pass = None , debug = <NUM_LIT:0> , <EOL> https_connection_factory = None , region = None , path = '<STR_LIT:/>' , <EOL> security_token = None , validate_certs = True , profile_name = None , <EOL> use_block_device_types = False ) : <EOL> """<STR_LIT>""" <EOL> if not region : <EOL> region = RegionInfo ( self , self . DefaultRegionName , <EOL> self . DefaultRegionEndpoint , <EOL> AutoScaleConnection ) <EOL> self . region = region <EOL> self . use_block_device_types = use_block_device_types <EOL> super ( AutoScaleConnection , self ) . __init__ ( aws_access_key_id , <EOL> aws_secret_access_key , <EOL> is_secure , port , proxy , proxy_port , <EOL> proxy_user , proxy_pass , <EOL> self . region . endpoint , debug , <EOL> https_connection_factory , path = path , <EOL> security_token = security_token , <EOL> validate_certs = validate_certs , <EOL> profile_name = profile_name ) <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def build_list_params ( self , params , items , label ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:1> , len ( items ) + <NUM_LIT:1> ) : <EOL> if isinstance ( items [ i - <NUM_LIT:1> ] , dict ) : <EOL> for k , v in six . iteritems ( items [ i - <NUM_LIT:1> ] ) : <EOL> if isinstance ( v , dict ) : <EOL> for kk , vv in six . iteritems ( v ) : <EOL> params [ '<STR_LIT>' % ( label , i , k , kk ) ] = vv <EOL> else : <EOL> params [ '<STR_LIT>' % ( label , i , k ) ] = v <EOL> elif isinstance ( items [ i - <NUM_LIT:1> ] , six . string_types ) : <EOL> params [ '<STR_LIT>' % ( label , i ) ] = items [ i - <NUM_LIT:1> ] <EOL> def _update_group ( self , op , as_group ) : <EOL> params = { '<STR_LIT>' : as_group . name , <EOL> '<STR_LIT>' : as_group . launch_config_name , <EOL> '<STR_LIT>' : as_group . min_size , <EOL> '<STR_LIT>' : as_group . max_size } <EOL> zones = as_group . availability_zones <EOL> self . build_list_params ( params , zones , '<STR_LIT>' ) <EOL> if as_group . desired_capacity is not None : <EOL> params [ '<STR_LIT>' ] = as_group . desired_capacity <EOL> if as_group . vpc_zone_identifier : <EOL> params [ '<STR_LIT>' ] = as_group . vpc_zone_identifier <EOL> if as_group . health_check_period : <EOL> params [ '<STR_LIT>' ] = as_group . health_check_period <EOL> if as_group . health_check_type : <EOL> params [ '<STR_LIT>' ] = as_group . health_check_type <EOL> if as_group . default_cooldown : <EOL> params [ '<STR_LIT>' ] = as_group . default_cooldown <EOL> if as_group . placement_group : <EOL> params [ '<STR_LIT>' ] = as_group . placement_group <EOL> if as_group . instance_id : <EOL> params [ '<STR_LIT>' ] = as_group . instance_id <EOL> if as_group . termination_policies : <EOL> self . build_list_params ( params , as_group . termination_policies , <EOL> '<STR_LIT>' ) <EOL> if op . startswith ( '<STR_LIT>' ) : <EOL> if as_group . load_balancers : <EOL> self . build_list_params ( params , as_group . load_balancers , <EOL> '<STR_LIT>' ) <EOL> if as_group . tags : <EOL> for i , tag in enumerate ( as_group . tags ) : <EOL> tag . build_params ( params , i + <NUM_LIT:1> ) <EOL> return self . get_object ( op , params , Request ) <EOL> def attach_instances ( self , name , instance_ids ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : name , <EOL> } <EOL> self . build_list_params ( params , instance_ids , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def detach_instances ( self , name , instance_ids , decrement_capacity = True ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : name } <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' if decrement_capacity else '<STR_LIT:false>' <EOL> self . build_list_params ( params , instance_ids , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def create_auto_scaling_group ( self , as_group ) : <EOL> """<STR_LIT>""" <EOL> return self . _update_group ( '<STR_LIT>' , as_group ) <EOL> def delete_auto_scaling_group ( self , name , force_delete = False ) : <EOL> """<STR_LIT>""" <EOL> if ( force_delete ) : <EOL> params = { '<STR_LIT>' : name , '<STR_LIT>' : '<STR_LIT:true>' } <EOL> else : <EOL> params = { '<STR_LIT>' : name } <EOL> return self . get_object ( '<STR_LIT>' , params , Request ) <EOL> def create_launch_configuration ( self , launch_config ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : launch_config . image_id , <EOL> '<STR_LIT>' : launch_config . name , <EOL> '<STR_LIT>' : launch_config . instance_type } <EOL> if launch_config . key_name : <EOL> params [ '<STR_LIT>' ] = launch_config . key_name <EOL> if launch_config . user_data : <EOL> user_data = launch_config . user_data <EOL> if isinstance ( user_data , six . text_type ) : <EOL> user_data = user_data . encode ( '<STR_LIT:utf-8>' ) <EOL> params [ '<STR_LIT>' ] = base64 . b64encode ( user_data ) . decode ( '<STR_LIT:utf-8>' ) <EOL> if launch_config . kernel_id : <EOL> params [ '<STR_LIT>' ] = launch_config . kernel_id <EOL> if launch_config . ramdisk_id : <EOL> params [ '<STR_LIT>' ] = launch_config . ramdisk_id <EOL> if launch_config . block_device_mappings : <EOL> [ x . autoscale_build_list_params ( params ) for x in launch_config . block_device_mappings ] <EOL> if launch_config . security_groups : <EOL> self . build_list_params ( params , launch_config . security_groups , <EOL> '<STR_LIT>' ) <EOL> if launch_config . instance_monitoring : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> else : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> if launch_config . spot_price is not None : <EOL> params [ '<STR_LIT>' ] = str ( launch_config . spot_price ) <EOL> if launch_config . instance_profile_name is not None : <EOL> params [ '<STR_LIT>' ] = launch_config . instance_profile_name <EOL> if launch_config . ebs_optimized : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> else : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> if launch_config . associate_public_ip_address is True : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> elif launch_config . associate_public_ip_address is False : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> if launch_config . volume_type : <EOL> params [ '<STR_LIT>' ] = launch_config . volume_type <EOL> if launch_config . delete_on_termination : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> else : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> if launch_config . iops : <EOL> params [ '<STR_LIT>' ] = launch_config . iops <EOL> if launch_config . classic_link_vpc_id : <EOL> params [ '<STR_LIT>' ] = launch_config . classic_link_vpc_id <EOL> if launch_config . classic_link_vpc_security_groups : <EOL> self . build_list_params ( <EOL> params , <EOL> launch_config . classic_link_vpc_security_groups , <EOL> '<STR_LIT>' <EOL> ) <EOL> return self . get_object ( '<STR_LIT>' , params , <EOL> Request , verb = '<STR_LIT:POST>' ) <EOL> def get_account_limits ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> return self . get_object ( '<STR_LIT>' , params , AccountLimits ) <EOL> def create_scaling_policy ( self , scaling_policy ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : scaling_policy . adjustment_type , <EOL> '<STR_LIT>' : scaling_policy . as_name , <EOL> '<STR_LIT>' : scaling_policy . name , <EOL> '<STR_LIT>' : scaling_policy . scaling_adjustment } <EOL> if scaling_policy . adjustment_type == "<STR_LIT>" and scaling_policy . min_adjustment_step is not None : <EOL> params [ '<STR_LIT>' ] = scaling_policy . min_adjustment_step <EOL> if scaling_policy . cooldown is not None : <EOL> params [ '<STR_LIT>' ] = scaling_policy . cooldown <EOL> return self . get_object ( '<STR_LIT>' , params , Request ) <EOL> def delete_launch_configuration ( self , launch_config_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : launch_config_name } <EOL> return self . get_object ( '<STR_LIT>' , params , Request ) <EOL> def get_all_groups ( self , names = None , max_records = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if names : <EOL> self . build_list_params ( params , names , '<STR_LIT>' ) <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , AutoScalingGroup ) ] ) <EOL> def get_all_launch_configurations ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> max_records = kwargs . get ( '<STR_LIT>' , None ) <EOL> names = kwargs . get ( '<STR_LIT>' , None ) <EOL> if max_records is not None : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if names : <EOL> self . build_list_params ( params , names , '<STR_LIT>' ) <EOL> next_token = kwargs . get ( '<STR_LIT>' ) <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , LaunchConfiguration ) ] ) <EOL> def get_all_activities ( self , autoscale_group , activity_ids = None , <EOL> max_records = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> name = autoscale_group <EOL> if isinstance ( autoscale_group , AutoScalingGroup ) : <EOL> name = autoscale_group . name <EOL> params = { '<STR_LIT>' : name } <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if activity_ids : <EOL> self . build_list_params ( params , activity_ids , '<STR_LIT>' ) <EOL> return self . get_list ( '<STR_LIT>' , <EOL> params , [ ( '<STR_LIT>' , Activity ) ] ) <EOL> def get_termination_policies ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_object ( '<STR_LIT>' , <EOL> { } , TerminationPolicies ) <EOL> def delete_scheduled_action ( self , scheduled_action_name , <EOL> autoscale_group = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : scheduled_action_name } <EOL> if autoscale_group : <EOL> params [ '<STR_LIT>' ] = autoscale_group <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def terminate_instance ( self , instance_id , decrement_capacity = True ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : instance_id } <EOL> if decrement_capacity : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> else : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> return self . get_object ( '<STR_LIT>' , params , <EOL> Activity ) <EOL> def delete_policy ( self , policy_name , autoscale_group = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : policy_name } <EOL> if autoscale_group : <EOL> params [ '<STR_LIT>' ] = autoscale_group <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def get_all_adjustment_types ( self ) : <EOL> return self . get_list ( '<STR_LIT>' , { } , <EOL> [ ( '<STR_LIT>' , AdjustmentType ) ] ) <EOL> def get_all_autoscaling_instances ( self , instance_ids = None , <EOL> max_records = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if instance_ids : <EOL> self . build_list_params ( params , instance_ids , '<STR_LIT>' ) <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , <EOL> params , [ ( '<STR_LIT>' , Instance ) ] ) <EOL> def get_all_metric_collection_types ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_object ( '<STR_LIT>' , <EOL> { } , MetricCollectionTypes ) <EOL> def get_all_policies ( self , as_group = None , policy_names = None , <EOL> max_records = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if as_group : <EOL> params [ '<STR_LIT>' ] = as_group <EOL> if policy_names : <EOL> self . build_list_params ( params , policy_names , '<STR_LIT>' ) <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , ScalingPolicy ) ] ) <EOL> def get_all_scaling_process_types ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_list ( '<STR_LIT>' , { } , <EOL> [ ( '<STR_LIT>' , ProcessType ) ] ) <EOL> def suspend_processes ( self , as_group , scaling_processes = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : as_group } <EOL> if scaling_processes : <EOL> self . build_list_params ( params , scaling_processes , <EOL> '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def resume_processes ( self , as_group , scaling_processes = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : as_group } <EOL> if scaling_processes : <EOL> self . build_list_params ( params , scaling_processes , <EOL> '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def create_scheduled_group_action ( self , as_group , name , time = None , <EOL> desired_capacity = None , <EOL> min_size = None , max_size = None , <EOL> start_time = None , end_time = None , <EOL> recurrence = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : as_group , <EOL> '<STR_LIT>' : name } <EOL> if start_time is not None : <EOL> params [ '<STR_LIT>' ] = start_time . isoformat ( ) <EOL> if end_time is not None : <EOL> params [ '<STR_LIT>' ] = end_time . isoformat ( ) <EOL> if recurrence is not None : <EOL> params [ '<STR_LIT>' ] = recurrence <EOL> if time : <EOL> params [ '<STR_LIT>' ] = time . isoformat ( ) <EOL> if desired_capacity is not None : <EOL> params [ '<STR_LIT>' ] = desired_capacity <EOL> if min_size is not None : <EOL> params [ '<STR_LIT>' ] = min_size <EOL> if max_size is not None : <EOL> params [ '<STR_LIT>' ] = max_size <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def get_all_scheduled_actions ( self , as_group = None , start_time = None , <EOL> end_time = None , scheduled_actions = None , <EOL> max_records = None , next_token = None ) : <EOL> params = { } <EOL> if as_group : <EOL> params [ '<STR_LIT>' ] = as_group <EOL> if scheduled_actions : <EOL> self . build_list_params ( params , scheduled_actions , <EOL> '<STR_LIT>' ) <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , ScheduledUpdateGroupAction ) ] ) <EOL> def disable_metrics_collection ( self , as_group , metrics = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : as_group } <EOL> if metrics : <EOL> self . build_list_params ( params , metrics , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def enable_metrics_collection ( self , as_group , granularity , metrics = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : as_group , <EOL> '<STR_LIT>' : granularity } <EOL> if metrics : <EOL> self . build_list_params ( params , metrics , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def execute_policy ( self , policy_name , as_group = None , honor_cooldown = None ) : <EOL> params = { '<STR_LIT>' : policy_name } <EOL> if as_group : <EOL> params [ '<STR_LIT>' ] = as_group <EOL> if honor_cooldown : <EOL> params [ '<STR_LIT>' ] = honor_cooldown <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def put_notification_configuration ( self , autoscale_group , topic , notification_types ) : <EOL> """<STR_LIT>""" <EOL> name = autoscale_group <EOL> if isinstance ( autoscale_group , AutoScalingGroup ) : <EOL> name = autoscale_group . name <EOL> params = { '<STR_LIT>' : name , <EOL> '<STR_LIT>' : topic } <EOL> self . build_list_params ( params , notification_types , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def delete_notification_configuration ( self , autoscale_group , topic ) : <EOL> """<STR_LIT>""" <EOL> name = autoscale_group <EOL> if isinstance ( autoscale_group , AutoScalingGroup ) : <EOL> name = autoscale_group . name <EOL> params = { '<STR_LIT>' : name , <EOL> '<STR_LIT>' : topic } <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def set_instance_health ( self , instance_id , health_status , <EOL> should_respect_grace_period = True ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : instance_id , <EOL> '<STR_LIT>' : health_status } <EOL> if should_respect_grace_period : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> else : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def set_desired_capacity ( self , group_name , desired_capacity , honor_cooldown = False ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : group_name , <EOL> '<STR_LIT>' : desired_capacity } <EOL> if honor_cooldown : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def get_all_tags ( self , filters = None , max_records = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , Tag ) ] ) <EOL> def create_or_update_tags ( self , tags ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> for i , tag in enumerate ( tags ) : <EOL> tag . build_params ( params , i + <NUM_LIT:1> ) <EOL> return self . get_status ( '<STR_LIT>' , params , verb = '<STR_LIT:POST>' ) <EOL> def delete_tags ( self , tags ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> for i , tag in enumerate ( tags ) : <EOL> tag . build_params ( params , i + <NUM_LIT:1> ) <EOL> return self . get_status ( '<STR_LIT>' , params , verb = '<STR_LIT:POST>' ) </s>
