Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def list(self, filter_by_owner=True, limit=None, marker=None):
"""
List all images.
"""
azure_images = self.provider.azure_client.list_images()
azure_gallery_refs = self.provider.azure_client.list_gallery_refs() \
if not filter_by_owner else []
cb_images = [AzureMachineImage(self.provider, img)
for img in azure_images + azure_gallery_refs]
return ClientPagedResultList(self.provider, cb_images,
limit=limit, marker=marker)
def list(self, firewall, limit=None, marker=None):
# Filter out firewall rules with priority < 3500 because values
# between 3500 and 4096 are assumed to be owned by cloudbridge
# default rules.
# pylint:disable=protected-access
rules = [AzureVMFirewallRule(firewall, rule) for rule
in firewall._vm_firewall.security_rules
if rule.priority < 3500]
return ClientPagedResultList(self.provider, rules,
limit=limit, marker=marker)
def find(self, bucket, **kwargs):
# pylint:disable=protected-access
obj_list = [AWSBucketObject(self.provider, o)
for o in bucket._bucket.objects.all()]
filters = ['name']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self.provider, list(matches),
limit=None, marker=None)
marker)
# Wrap in CB objects.
results = [self.cb_resource(self.provider, obj) for obj in boto_objs]
if pag_type == 'server':
log.debug("Using server pagination.")
return ServerPagedResultList(is_truncated=True if resume_token
else False,
marker=resume_token if resume_token
else None,
supports_total=False,
data=results)
else:
log.debug("Did not received a resume token, will page in client"
" if necessary.")
return ClientPagedResultList(self.provider, results,
limit=limit, marker=marker)
def list(self, limit=None, marker=None):
routes = [AzureRouter(self.provider, route)
for route in
self.provider.azure_client.list_route_tables()]
return ClientPagedResultList(self.provider,
routes,
limit=limit, marker=marker)
def list(self, limit=None, marker=None):
regions = [AzureRegion(self.provider, region)
for region in self.provider.azure_client.list_locations()]
return ClientPagedResultList(self.provider, regions,
limit=limit, marker=marker)
def list(self, limit=None, marker=None):
azure_vols = self.provider.azure_client.list_disks()
cb_vols = [AzureVolume(self.provider, vol) for vol in azure_vols]
return ClientPagedResultList(self.provider, cb_vols,
limit=limit, marker=marker)
def list(self, limit=None, marker=None):
fws = [AzureVMFirewall(self.provider, fw)
for fw in self.provider.azure_client.list_vm_firewall()]
return ClientPagedResultList(self.provider, fws, limit, marker)
def list(self, limit=None, marker=None):
buckets = [AzureBucket(self.provider, bucket)
for bucket
in self.provider.azure_client.list_containers()[0]]
return ClientPagedResultList(self.provider, buckets,
limit=limit, marker=marker)
def list(self, limit=None, marker=None):
vm_types = [AWSVMType(self.provider, vm_type)
for vm_type in self.instance_data]
return ClientPagedResultList(self.provider, vm_types,
limit=limit, marker=marker)