Source code for boto3.resources.model

# 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.

"""
The models defined in this file represent the resource JSON description
format and provide a layer of abstraction from the raw JSON. The advantages
of this are:

* Pythonic interface (e.g. ``action.request.operation``)
* Consumers need not change for minor JSON changes (e.g. renamed field)

These models are used both by the resource factory to generate resource
classes as well as by the documentation generator.
"""

import logging

from botocore import xform_name

logger = logging.getLogger(__name__)


[docs]class Identifier: """ A resource identifier, given by its name. :type name: string :param name: The name of the identifier """ def __init__(self, name, member_name=None): #: (``string``) The name of the identifier self.name = name self.member_name = member_name
[docs]class Action: """ A service operation action. :type name: string :param name: The name of the action :type definition: dict :param definition: The JSON definition :type resource_defs: dict :param resource_defs: All resources defined in the service """ def __init__(self, name, definition, resource_defs): self._definition = definition #: (``string``) The name of the action self.name = name #: (:py:class:`Request`) This action's request or ``None`` self.request = None if 'request' in definition: self.request = Request(definition.get('request', {})) #: (:py:class:`ResponseResource`) This action's resource or ``None`` self.resource = None if 'resource' in definition: self.resource = ResponseResource( definition.get('resource', {}), resource_defs ) #: (``string``) The JMESPath search path or ``None`` self.path = definition.get('path')
[docs]class DefinitionWithParams: """ An item which has parameters exposed via the ``params`` property. A request has an operation and parameters, while a waiter has a name, a low-level waiter name and parameters. :type definition: dict :param definition: The JSON definition """ def __init__(self, definition): self._definition = definition @property def params(self): """ Get a list of auto-filled parameters for this request. :type: list(:py:class:`Parameter`) """ params = [] for item in self._definition.get('params', []): params.append(Parameter(**item)) return params
[docs]class Parameter: """ An auto-filled parameter which has a source and target. For example, the ``QueueUrl`` may be auto-filled from a resource's ``url`` identifier when making calls to ``queue.receive_messages``. :type target: string :param target: The destination parameter name, e.g. ``QueueUrl`` :type source_type: string :param source_type: Where the source is defined. :type source: string :param source: The source name, e.g. ``Url`` """ def __init__( self, target, source, name=None, path=None, value=None, **kwargs ): #: (``string``) The destination parameter name self.target = target #: (``string``) Where the source is defined self.source = source #: (``string``) The name of the source, if given self.name = name #: (``string``) The JMESPath query of the source self.path = path #: (``string|int|float|bool``) The source constant value self.value = value # Complain if we encounter any unknown values. if kwargs: logger.warning('Unknown parameter options found: %s', kwargs)
[docs]class Request(DefinitionWithParams): """ A service operation action request. :type definition: dict :param definition: The JSON definition """ def __init__(self, definition): super().__init__(definition) #: (``string``) The name of the low-level service operation self.operation = definition.get('operation')
[docs]class Waiter(DefinitionWithParams): """ An event waiter specification. :type name: string :param name: Name of the waiter :type definition: dict :param definition: The JSON definition """ PREFIX = 'WaitUntil' def __init__(self, name, definition): super().__init__(definition) #: (``string``) The name of this waiter self.name = name #: (``string``) The name of the underlying event waiter self.waiter_name = definition.get('waiterName')
[docs]class ResponseResource: """ A resource response to create after performing an action. :type definition: dict :param definition: The JSON definition :type resource_defs: dict :param resource_defs: All resources defined in the service """ def __init__(self, definition, resource_defs): self._definition = definition self._resource_defs = resource_defs #: (``string``) The name of the response resource type self.type = definition.get('type') #: (``string``) The JMESPath search query or ``None`` self.path = definition.get('path') @property def identifiers(self): """ A list of resource identifiers. :type: list(:py:class:`Identifier`) """ identifiers = [] for item in self._definition.get('identifiers', []): identifiers.append(Parameter(**item)) return identifiers @property def model(self): """ Get the resource model for the response resource. :type: :py:class:`ResourceModel` """ return ResourceModel( self.type, self._resource_defs[self.type], self._resource_defs )
[docs]class Collection(Action): """ A group of resources. See :py:class:`Action`. :type name: string :param name: The name of the collection :type definition: dict :param definition: The JSON definition :type resource_defs: dict :param resource_defs: All resources defined in the service """ @property def batch_actions(self): """ Get a list of batch actions supported by the resource type contained in this action. This is a shortcut for accessing the same information through the resource model. :rtype: list(:py:class:`Action`) """ return self.resource.model.batch_actions
[docs]class ResourceModel: """ A model representing a resource, defined via a JSON description format. A resource has identifiers, attributes, actions, sub-resources, references and collections. For more information on resources, see :ref:`guide_resources`. :type name: string :param name: The name of this resource, e.g. ``sqs`` or ``Queue`` :type definition: dict :param definition: The JSON definition :type resource_defs: dict :param resource_defs: All resources defined in the service """ def __init__(self, name, definition, resource_defs): self._definition = definition self._resource_defs = resource_defs self._renamed = {} #: (``string``) The name of this resource self.name = name #: (``string``) The service shape name for this resource or ``None`` self.shape = definition.get('shape')
[docs] def load_rename_map(self, shape=None): """ Load a name translation map given a shape. This will set up renamed values for any collisions, e.g. if the shape, an action, and a subresource all are all named ``foo`` then the resource will have an action ``foo``, a subresource named ``Foo`` and a property named ``foo_attribute``. This is the order of precedence, from most important to least important: * Load action (resource.load) * Identifiers * Actions * Subresources * References * Collections * Waiters * Attributes (shape members) Batch actions are only exposed on collections, so do not get modified here. Subresources use upper camel casing, so are unlikely to collide with anything but other subresources. Creates a structure like this:: renames = { ('action', 'id'): 'id_action', ('collection', 'id'): 'id_collection', ('attribute', 'id'): 'id_attribute' } # Get the final name for an action named 'id' name = renames.get(('action', 'id'), 'id') :type shape: botocore.model.Shape :param shape: The underlying shape for this resource. """ # Meta is a reserved name for resources names = {'meta'} self._renamed = {} if self._definition.get('load'): names.add('load') for item in self._definition.get('identifiers', []): self._load_name_with_category(names, item['name'], 'identifier') for name in self._definition.get('actions', {}): self._load_name_with_category(names, name, 'action') for name, ref in self._get_has_definition().items(): # Subresources require no data members, just typically # identifiers and user input. data_required = False for identifier in ref['resource']['identifiers']: if identifier['source'] == 'data': data_required = True break if not data_required: self._load_name_with_category( names, name, 'subresource', snake_case=False ) else: self._load_name_with_category(names, name, 'reference') for name in self._definition.get('hasMany', {}): self._load_name_with_category(names, name, 'collection') for name in self._definition.get('waiters', {}): self._load_name_with_category( names, Waiter.PREFIX + name, 'waiter' ) if shape is not None: for name in shape.members.keys(): self._load_name_with_category(names, name, 'attribute')
def _load_name_with_category(self, names, name, category, snake_case=True): """ Load a name with a given category, possibly renaming it if that name is already in use. The name will be stored in ``names`` and possibly be set up in ``self._renamed``. :type names: set :param names: Existing names (Python attributes, properties, or methods) on the resource. :type name: string :param name: The original name of the value. :type category: string :param category: The value type, such as 'identifier' or 'action' :type snake_case: bool :param snake_case: True (default) if the name should be snake cased. """ if snake_case: name = xform_name(name) if name in names: logger.debug(f'Renaming {self.name} {category} {name}') self._renamed[(category, name)] = name + '_' + category name += '_' + category if name in names: # This isn't good, let's raise instead of trying to keep # renaming this value. raise ValueError( f'Problem renaming {self.name} {category} to {name}!' ) names.add(name) def _get_name(self, category, name, snake_case=True): """ Get a possibly renamed value given a category and name. This uses the rename map set up in ``load_rename_map``, so that method must be called once first. :type category: string :param category: The value type, such as 'identifier' or 'action' :type name: string :param name: The original name of the value :type snake_case: bool :param snake_case: True (default) if the name should be snake cased. :rtype: string :return: Either the renamed value if it is set, otherwise the original name. """ if snake_case: name = xform_name(name) return self._renamed.get((category, name), name)
[docs] def get_attributes(self, shape): """ Get a dictionary of attribute names to original name and shape models that represent the attributes of this resource. Looks like the following: { 'some_name': ('SomeName', <Shape...>) } :type shape: botocore.model.Shape :param shape: The underlying shape for this resource. :rtype: dict :return: Mapping of resource attributes. """ attributes = {} identifier_names = [i.name for i in self.identifiers] for name, member in shape.members.items(): snake_cased = xform_name(name) if snake_cased in identifier_names: # Skip identifiers, these are set through other means continue snake_cased = self._get_name( 'attribute', snake_cased, snake_case=False ) attributes[snake_cased] = (name, member) return attributes
@property def identifiers(self): """ Get a list of resource identifiers. :type: list(:py:class:`Identifier`) """ identifiers = [] for item in self._definition.get('identifiers', []): name = self._get_name('identifier', item['name']) member_name = item.get('memberName', None) if member_name: member_name = self._get_name('attribute', member_name) identifiers.append(Identifier(name, member_name)) return identifiers @property def load(self): """ Get the load action for this resource, if it is defined. :type: :py:class:`Action` or ``None`` """ action = self._definition.get('load') if action is not None: action = Action('load', action, self._resource_defs) return action @property def actions(self): """ Get a list of actions for this resource. :type: list(:py:class:`Action`) """ actions = [] for name, item in self._definition.get('actions', {}).items(): name = self._get_name('action', name) actions.append(Action(name, item, self._resource_defs)) return actions @property def batch_actions(self): """ Get a list of batch actions for this resource. :type: list(:py:class:`Action`) """ actions = [] for name, item in self._definition.get('batchActions', {}).items(): name = self._get_name('batch_action', name) actions.append(Action(name, item, self._resource_defs)) return actions def _get_has_definition(self): """ Get a ``has`` relationship definition from a model, where the service resource model is treated special in that it contains a relationship to every resource defined for the service. This allows things like ``s3.Object('bucket-name', 'key')`` to work even though the JSON doesn't define it explicitly. :rtype: dict :return: Mapping of names to subresource and reference definitions. """ if self.name not in self._resource_defs: # This is the service resource, so let us expose all of # the defined resources as subresources. definition = {} for name, resource_def in self._resource_defs.items(): # It's possible for the service to have renamed a # resource or to have defined multiple names that # point to the same resource type, so we need to # take that into account. found = False has_items = self._definition.get('has', {}).items() for has_name, has_def in has_items: if has_def.get('resource', {}).get('type') == name: definition[has_name] = has_def found = True if not found: # Create a relationship definition and attach it # to the model, such that all identifiers must be # supplied by the user. It will look something like: # # { # 'resource': { # 'type': 'ResourceName', # 'identifiers': [ # {'target': 'Name1', 'source': 'input'}, # {'target': 'Name2', 'source': 'input'}, # ... # ] # } # } # fake_has = {'resource': {'type': name, 'identifiers': []}} for identifier in resource_def.get('identifiers', []): fake_has['resource']['identifiers'].append( {'target': identifier['name'], 'source': 'input'} ) definition[name] = fake_has else: definition = self._definition.get('has', {}) return definition def _get_related_resources(self, subresources): """ Get a list of sub-resources or references. :type subresources: bool :param subresources: ``True`` to get sub-resources, ``False`` to get references. :rtype: list(:py:class:`Action`) """ resources = [] for name, definition in self._get_has_definition().items(): if subresources: name = self._get_name('subresource', name, snake_case=False) else: name = self._get_name('reference', name) action = Action(name, definition, self._resource_defs) data_required = False for identifier in action.resource.identifiers: if identifier.source == 'data': data_required = True break if subresources and not data_required: resources.append(action) elif not subresources and data_required: resources.append(action) return resources @property def subresources(self): """ Get a list of sub-resources. :type: list(:py:class:`Action`) """ return self._get_related_resources(True) @property def references(self): """ Get a list of reference resources. :type: list(:py:class:`Action`) """ return self._get_related_resources(False) @property def collections(self): """ Get a list of collections for this resource. :type: list(:py:class:`Collection`) """ collections = [] for name, item in self._definition.get('hasMany', {}).items(): name = self._get_name('collection', name) collections.append(Collection(name, item, self._resource_defs)) return collections @property def waiters(self): """ Get a list of waiters for this resource. :type: list(:py:class:`Waiter`) """ waiters = [] for name, item in self._definition.get('waiters', {}).items(): name = self._get_name('waiter', Waiter.PREFIX + name) waiters.append(Waiter(name, item)) return waiters