Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
all_clusters_instances = list(
ec2.instances.filter(
Filters=[
{'Name': 'instance-state-name', 'Values': ['pending', 'running', 'stopping', 'stopped']},
{'Name': 'instance.group-name', 'Values': group_name_filter},
{'Name': 'vpc-id', 'Values': [vpc_id]},
]))
found_cluster_names = {
_get_cluster_name(instance) for instance in all_clusters_instances}
if cluster_names:
missing_cluster_names = set(cluster_names) - found_cluster_names
if missing_cluster_names:
raise ClusterNotFound("No cluster {c} in region {r}.".format(
c=missing_cluster_names.pop(),
r=region))
clusters = [
_compose_cluster(
name=cluster_name,
region=region,
vpc_id=vpc_id,
instances=list(filter(
lambda x: _get_cluster_name(x) == cluster_name, all_clusters_instances)))
for cluster_name in found_cluster_names]
return clusters
if not vpc_id:
vpc_id = get_default_vpc(region=region).id
else:
# If it's a non-default VPC -- i.e. the user set it up -- make sure it's
# configured correctly.
check_network_config(
region_name=region,
vpc_id=vpc_id,
subnet_id=subnet_id)
try:
get_cluster(
cluster_name=cluster_name,
region=region,
vpc_id=vpc_id)
except ClusterNotFound as e:
pass
else:
raise ClusterAlreadyExists(
"Cluster {c} already exists in region {r}, VPC {v}.".format(
c=cluster_name,
r=region,
v=vpc_id))
flintrock_security_groups = get_or_create_flintrock_security_groups(
cluster_name=cluster_name,
vpc_id=vpc_id,
region=region)
user_security_groups = get_security_groups(
vpc_id=vpc_id,
region=region,
security_group_names=security_groups)