# 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