def clientWrapper(service, region=None):
if region:
cl = boto3.client(service, region_name=region)
else:
try:
cl = boto3.client(service)
except NoRegionError:
print("You need to specify a region in some way. See "
"http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-"
"getting-started.html for ways that you can set it "
"globally, or pass 'region' as an argument to this task.")
raise SystemExit(1)
return cl
# Default namespace
python类NoRegionError()的实例源码
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
def test_get_resource_model_attributes(self):
try:
resource = boto3.resource('ec2')
except NoRegionError:
# skip for environment that doesn't have boto config like CI
pass
else:
collection = resource.instances.all()
attributes = get_resource_model_attributes(resource, collection)
assert attributes
assert 'instance_id' in attributes
assert 'image_id' in attributes
def main(args=None):
parsed_args = parser.parse_args(args=args)
logger.setLevel(parsed_args.log_level)
has_attrs = (getattr(parsed_args, "sort_by", None) and
getattr(parsed_args, "columns", None))
if has_attrs and parsed_args.sort_by not in parsed_args.columns:
parsed_args.columns.append(parsed_args.sort_by)
try:
result = parsed_args.entry_point(parsed_args)
except Exception as e:
if isinstance(e, NoRegionError):
msg = "The AWS CLI is not configured."
msg += " Please configure it using instructions at"
msg += " http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html"
exit(msg)
elif logger.level < logging.ERROR:
raise
else:
err_msg = traceback.format_exc()
try:
err_log_filename = os.path.join(config.user_config_dir, "error.log")
with open(err_log_filename, "ab") as fh:
print(datetime.datetime.now().isoformat(), file=fh)
print(err_msg, file=fh)
exit("{}: {}. See {} for error details.".format(e.__class__.__name__, e, err_log_filename))
except Exception:
print(err_msg, file=sys.stderr)
exit(os.EX_SOFTWARE)
if isinstance(result, SystemExit):
raise result
elif result is not None:
if isinstance(result, dict) and "ResponseMetadata" in result:
del result["ResponseMetadata"]
print(json.dumps(result, indent=2, default=lambda x: str(x)))
def kms_client(self):
if not hasattr(self, '_kms_client'):
try:
self._kms_client = boto3.client('kms')
except NoRegionError:
region_name = get_current_region()
self._kms_client = boto3.client('kms', region_name=region_name)
return self._kms_client
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
regions.py 文件源码
项目：tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 30
收藏 0
点赞 0
评论 0
def _endpoint_for_partition(self, partition, service_name, region_name):
# Get the service from the partition, or an empty template.
service_data = partition['services'].get(
service_name, DEFAULT_SERVICE_DATA)
# Use the partition endpoint if no region is supplied.
if region_name is None:
if 'partitionEndpoint' in service_data:
region_name = service_data['partitionEndpoint']
else:
raise NoRegionError()
# Attempt to resolve the exact region for this partition.
if region_name in service_data['endpoints']:
return self._resolve(
partition, service_name, service_data, region_name)
# Check to see if the endpoint provided is valid for the partition.
if self._region_match(partition, region_name):
# Use the partition endpoint if set and not regionalized.
partition_endpoint = service_data.get('partitionEndpoint')
is_regionalized = service_data.get('isRegionalized', True)
if partition_endpoint and not is_regionalized:
LOG.debug('Using partition endpoint for %s, %s: %s',
service_name, region_name, partition_endpoint)
return self._resolve(
partition, service_name, service_data, partition_endpoint)
LOG.debug('Creating a regex based endpoint for %s, %s',
service_name, region_name)
return self._resolve(
partition, service_name, service_data, region_name)
def main(self, args=None):
"""
:param args: List of arguments, with the 'aws' removed. For example,
the command "aws s3 list-objects --bucket foo" will have an
args list of ``['s3', 'list-objects', '--bucket', 'foo']``.
"""
if args is None:
args = sys.argv[1:]
command_table = self._get_command_table()
parser = self._create_parser(command_table)
self._add_aliases(command_table, parser)
parsed_args, remaining = parser.parse_known_args(args)
try:
# Because _handle_top_level_args emits events, it's possible
# that exceptions can be raised, which should have the same
# general exception handling logic as calling into the
# command table. This is why it's in the try/except clause.
self._handle_top_level_args(parsed_args)
self._emit_session_event()
return command_table[parsed_args.command](remaining, parsed_args)
except UnknownArgumentError as e:
sys.stderr.write("usage: %s\n" % USAGE)
sys.stderr.write(str(e))
sys.stderr.write("\n")
return 255
except NoRegionError as e:
msg = ('%s You can also configure your region by running '
'"aws configure".' % e)
self._show_error(msg)
return 255
except NoCredentialsError as e:
msg = ('%s. You can configure credentials by running '
'"aws configure".' % e)
self._show_error(msg)
return 255
except KeyboardInterrupt:
# Shell standard for signals that terminate
# the process is to return 128 + signum, in this case
# SIGINT=2, so we'll have an RC of 130.
sys.stdout.write("\n")
return 128 + signal.SIGINT
except Exception as e:
LOG.debug("Exception caught in main()", exc_info=True)
LOG.debug("Exiting with rc 255")
err = get_stderr_text_writer()
err.write("\n")
err.write(six.text_type(e))
err.write("\n")
return 255
