Source code for boto3.resources.base

# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# https://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.

import logging

import boto3

logger = logging.getLogger(__name__)


[docs]class ResourceMeta: """ An object containing metadata about a resource. """ def __init__( self, service_name, identifiers=None, client=None, data=None, resource_model=None, ): #: (``string``) The service name, e.g. 's3' self.service_name = service_name if identifiers is None: identifiers = [] #: (``list``) List of identifier names self.identifiers = identifiers #: (:py:class:`~botocore.client.BaseClient`) Low-level Botocore client self.client = client #: (``dict``) Loaded resource data attributes self.data = data # The resource model for that resource self.resource_model = resource_model def __repr__(self): return f'ResourceMeta(\'{self.service_name}\', identifiers={self.identifiers})' def __eq__(self, other): # Two metas are equal if their components are all equal if other.__class__.__name__ != self.__class__.__name__: return False return self.__dict__ == other.__dict__
[docs] def copy(self): """ Create a copy of this metadata object. """ params = self.__dict__.copy() service_name = params.pop('service_name') return ResourceMeta(service_name, **params)
[docs]class ServiceResource: """ A base class for resources. :type client: botocore.client :param client: A low-level Botocore client instance """ meta = None """ Stores metadata about this resource instance, such as the ``service_name``, the low-level ``client`` and any cached ``data`` from when the instance was hydrated. For example:: # Get a low-level client from a resource instance client = resource.meta.client response = client.operation(Param='foo') # Print the resource instance's service short name print(resource.meta.service_name) See :py:class:`ResourceMeta` for more information. """ def __init__(self, *args, **kwargs): # Always work on a copy of meta, otherwise we would affect other # instances of the same subclass. self.meta = self.meta.copy() # Create a default client if none was passed if kwargs.get('client') is not None: self.meta.client = kwargs.get('client') else: self.meta.client = boto3.client(self.meta.service_name) # Allow setting identifiers as positional arguments in the order # in which they were defined in the ResourceJSON. for i, value in enumerate(args): setattr(self, '_' + self.meta.identifiers[i], value) # Allow setting identifiers via keyword arguments. Here we need # extra logic to ignore other keyword arguments like ``client``. for name, value in kwargs.items(): if name == 'client': continue if name not in self.meta.identifiers: raise ValueError(f'Unknown keyword argument: {name}') setattr(self, '_' + name, value) # Validate that all identifiers have been set. for identifier in self.meta.identifiers: if getattr(self, identifier) is None: raise ValueError(f'Required parameter {identifier} not set') def __repr__(self): identifiers = [] for identifier in self.meta.identifiers: identifiers.append( f'{identifier}={repr(getattr(self, identifier))}' ) return "{}({})".format( self.__class__.__name__, ', '.join(identifiers), ) def __eq__(self, other): # Should be instances of the same resource class if other.__class__.__name__ != self.__class__.__name__: return False # Each of the identifiers should have the same value in both # instances, e.g. two buckets need the same name to be equal. for identifier in self.meta.identifiers: if getattr(self, identifier) != getattr(other, identifier): return False return True def __hash__(self): identifiers = [] for identifier in self.meta.identifiers: identifiers.append(getattr(self, identifier)) return hash((self.__class__.__name__, tuple(identifiers)))