Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if matches:
return matches[0]
else:
return None
@dispatch(event="security.vm_firewall_rules.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, firewall, **kwargs):
obj_list = firewall.rules
filters = ['name', 'direction', 'protocol', 'from_port', 'to_port',
'cidr', 'src_dest_fw', 'src_dest_fw_id']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseStorageService(StorageService, BaseCloudService):
def __init__(self, provider):
super(BaseStorageService, self).__init__(provider)
class BaseVolumeService(
BasePageableObjectMixin, VolumeService, BaseCloudService):
def __init__(self, provider):
super(BaseVolumeService, self).__init__(provider)
self._service_event_pattern += ".storage.volumes"
class BaseSnapshotService(
BasePageableObjectMixin, SnapshotService, BaseCloudService):
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def get(self, vm_type_id):
vm_type = (t for t in self if t.id == vm_type_id)
return next(vm_type, None)
@dispatch(event="compute.vm_types.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, **kwargs):
obj_list = self
filters = ['name']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseRegionService(
BasePageableObjectMixin, RegionService, BaseCloudService):
def __init__(self, provider):
super(BaseRegionService, self).__init__(provider)
self._service_event_pattern += ".compute.regions"
@dispatch(event="compute.regions.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, **kwargs):
obj_list = self
filters = ['name']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseNetworkingService(NetworkingService, BaseCloudService):
def find(self, firewall, **kwargs):
obj_list = firewall.rules
filters = ['name', 'direction', 'protocol', 'from_port', 'to_port',
'cidr', 'src_dest_fw', 'src_dest_fw_id']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseStorageService(StorageService, BaseCloudService):
def __init__(self, provider):
super(BaseStorageService, self).__init__(provider)
class BaseVolumeService(
BasePageableObjectMixin, VolumeService, BaseCloudService):
def __init__(self, provider):
super(BaseVolumeService, self).__init__(provider)
self._service_event_pattern += ".storage.volumes"
class BaseSnapshotService(
BasePageableObjectMixin, SnapshotService, BaseCloudService):
def __init__(self, provider):
super(BaseSnapshotService, self).__init__(provider)
self._service_event_pattern += ".storage.snapshots"
class BaseBucketService(
BasePageableObjectMixin, BucketService, BaseCloudService):
class BaseStorageService(StorageService, BaseCloudService):
def __init__(self, provider):
super(BaseStorageService, self).__init__(provider)
class BaseVolumeService(
BasePageableObjectMixin, VolumeService, BaseCloudService):
def __init__(self, provider):
super(BaseVolumeService, self).__init__(provider)
self._service_event_pattern += ".storage.volumes"
class BaseSnapshotService(
BasePageableObjectMixin, SnapshotService, BaseCloudService):
def __init__(self, provider):
super(BaseSnapshotService, self).__init__(provider)
self._service_event_pattern += ".storage.snapshots"
class BaseBucketService(
BasePageableObjectMixin, BucketService, BaseCloudService):
def __init__(self, provider):
super(BaseBucketService, self).__init__(provider)
self._service_event_pattern += ".storage.buckets"
# Generic find will be used for providers where we have not implemented
# provider-specific querying for find method
@dispatch(event="storage.buckets.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, **kwargs):
obj_list = self
filters = ['name']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseNetworkingService(NetworkingService, BaseCloudService):
def __init__(self, provider):
super(BaseNetworkingService, self).__init__(provider)
class BaseNetworkService(
BasePageableObjectMixin, NetworkService, BaseCloudService):
def __init__(self, provider):
super(BaseNetworkService, self).__init__(provider)
self._service_event_pattern += ".networking.networks"
@property
def subnets(self):
return [subnet for subnet in self.provider.subnets
if subnet.network_id == self.id]
def get_or_create_default(self):
networks = self.provider.networking.networks.find(
label=BaseNetwork.CB_DEFAULT_NETWORK_LABEL)
if networks:
return networks[0]
BasePageableObjectMixin, RegionService, BaseCloudService):
def __init__(self, provider):
super(BaseRegionService, self).__init__(provider)
self._service_event_pattern += ".compute.regions"
@dispatch(event="compute.regions.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, **kwargs):
obj_list = self
filters = ['name']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))
class BaseNetworkingService(NetworkingService, BaseCloudService):
def __init__(self, provider):
super(BaseNetworkingService, self).__init__(provider)
class BaseNetworkService(
BasePageableObjectMixin, NetworkService, BaseCloudService):
def __init__(self, provider):
super(BaseNetworkService, self).__init__(provider)
self._service_event_pattern += ".networking.networks"
@property
def subnets(self):
return [subnet for subnet in self.provider.subnets
if subnet.network_id == self.id]
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def get(self, firewall, rule_id):
matches = [rule for rule in firewall.rules if rule.id == rule_id]
if matches:
return matches[0]
else:
return None
class BaseComputeService(ComputeService, BaseCloudService):
def __init__(self, provider):
super(BaseComputeService, self).__init__(provider)
class BaseImageService(
BasePageableObjectMixin, ImageService, BaseCloudService):
def __init__(self, provider):
super(BaseImageService, self).__init__(provider)
self._service_event_pattern += ".compute.images"
class BaseInstanceService(
BasePageableObjectMixin, InstanceService, BaseCloudService):
def __init__(self, provider):
super(BaseInstanceService, self).__init__(provider)
self._service_event_pattern += ".compute.instances"
class BaseVMTypeService(
BasePageableObjectMixin, VMTypeService, BaseCloudService):
def __init__(self, provider):
super(BaseVMTypeService, self).__init__(provider)
self._service_event_pattern += ".compute.vm_types"
@dispatch(event="compute.vm_types.get",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def get(self, vm_type_id):
def provider(self):
return self._provider
@property
def events(self):
return self._provider.middleware.events
class BaseSecurityService(SecurityService, BaseCloudService):
def __init__(self, provider):
super(BaseSecurityService, self).__init__(provider)
class BaseKeyPairService(
BasePageableObjectMixin, KeyPairService, BaseCloudService):
def __init__(self, provider):
super(BaseKeyPairService, self).__init__(provider)
self._service_event_pattern += ".security.key_pairs"
class BaseVMFirewallService(
BasePageableObjectMixin, VMFirewallService, BaseCloudService):
def __init__(self, provider):
super(BaseVMFirewallService, self).__init__(provider)
self._service_event_pattern += ".security.vm_firewalls"
@dispatch(event="security.vm_firewalls.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, **kwargs):
return self.provider.networking.routers.create(
network=net_id, label=BaseRouter.CB_DEFAULT_ROUTER_LABEL)
class BaseGatewayService(GatewayService, BaseCloudService):
def __init__(self, provider):
super(BaseGatewayService, self).__init__(provider)
self._provider = provider
@property
def provider(self):
return self._provider
class BaseFloatingIPService(FloatingIPService, BaseCloudService):
def __init__(self, provider):
super(BaseFloatingIPService, self).__init__(provider)
self._provider = provider
@property
def provider(self):
return self._provider
@dispatch(event="networking.floating_ips.find",
priority=BaseCloudService.STANDARD_EVENT_PRIORITY)
def find(self, gateway, **kwargs):
obj_list = gateway.floating_ips
filters = ['name', 'public_ip']
matches = cb_helpers.generic_find(filters, kwargs, obj_list)
return ClientPagedResultList(self._provider, list(matches))