Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py	0 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 action.py # 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 from botocore import xform_name from boto3.docs.docstring import ActionDocstring from boto3.utils import inject_attribute from .model import Action from .params import create_request_parameters from .response import RawHandler, ResourceHandler logger = logging.getLogger(__name__) class ServiceAction: """ A class representing a callable action on a resource, for example ``sqs.get_queue_by_name(...)`` or ``s3.Bucket('foo').delete()``. The action may construct parameters from existing resource identifiers and may return either a raw response or a new resource instance. :type action_model: :py:class`~boto3.resources.model.Action` :param action_model: The action model. :type factory: ResourceFactory :param factory: The factory that created the resource class to which this action is attached. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service """ def __init__(self, action_model, factory=None, service_context=None): self._action_model = action_model # In the simplest case we just return the response, but if a # resource is defined, then we must create these before returning. resource_response_model = action_model.resource if resource_response_model: self._response_handler = ResourceHandler( search_path=resource_response_model.path, factory=factory, resource_model=resource_response_model, service_context=service_context, operation_name=action_model.request.operation, ) else: self._response_handler = RawHandler(action_model.path) def __call__(self, parent, args, kwargs): """ Perform the action's request operation after building operation parameters and build any defined resources from the response. :type parent: :py:class:`~boto3.resources.base.ServiceResource` :param parent: The resource instance to which this action is attached. :rtype: dict or ServiceResource or list(ServiceResource) :return: The response, either as a raw dict or resource instance(s). """ operation_name = xform_name(self._action_model.request.operation) # First, build predefined params and then update with the # user-supplied kwargs, which allows overriding the pre-built # params if needed. params = create_request_parameters(parent, self._action_model.request) params.update(kwargs) logger.debug( 'Calling %s:%s with %r', parent.meta.service_name, operation_name, params, ) response = getattr(parent.meta.client, operation_name)(args, *params) logger.debug('Response: %r', response) return self._response_handler(parent, params, response) class BatchAction(ServiceAction): """ An action which operates on a batch of items in a collection, typically a single page of results from the collection's underlying service operation call. For example, this allows you to delete up to 999 S3 objects in a single operation rather than calling ``.delete()`` on each one individually. :type action_model: :py:class`~boto3.resources.model.Action` :param action_model: The action model. :type factory: ResourceFactory :param factory: The factory that created the resource class to which this action is attached. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service """ def __call__(self, parent, args, *kwargs): """ Perform the batch action's operation on every page of results from the collection. :type parent: :py:class:`~boto3.resources.collection.ResourceCollection` :param parent: The collection iterator to which this action is attached. :rtype: list(dict) :return: A list of low-level response dicts from each call. """ service_name = None client = None responses = [] operation_name = xform_name(self._action_model.request.operation) # Unlike the simple action above, a batch action must operate # on batches (or pages) of items. So we get each page, construct # the necessary parameters and call the batch operation. for page in parent.pages(): params = {} for index, resource in enumerate(page): # There is no public interface to get a service name # or low-level client from a collection, so we get # these from the first resource in the collection. if service_name is None: service_name = resource.meta.service_name if client is None: client = resource.meta.client create_request_parameters( resource, self._action_model.request, params=params, index=index, ) if not params: # There are no items, no need to make a call. break params.update(kwargs) logger.debug( 'Calling %s:%s with %r', service_name, operation_name, params ) response = getattr(client, operation_name)(args, *params) logger.debug('Response: %r', response) responses.append(self._response_handler(parent, params, response)) return responses class WaiterAction: """ A class representing a callable waiter action on a resource, for example ``s3.Bucket('foo').wait_until_bucket_exists()``. The waiter action may construct parameters from existing resource identifiers. :type waiter_model: :py:class`~boto3.resources.model.Waiter` :param waiter_model: The action waiter. :type waiter_resource_name: string :param waiter_resource_name: The name of the waiter action for the resource. It usually begins with a ``wait_until_`` """ def __init__(self, waiter_model, waiter_resource_name): self._waiter_model = waiter_model self._waiter_resource_name = waiter_resource_name def __call__(self, parent, args, kwargs): """ Perform the wait operation after building operation parameters. :type parent: :py:class:`~boto3.resources.base.ServiceResource` :param parent: The resource instance to which this action is attached. """ client_waiter_name = xform_name(self._waiter_model.waiter_name) # First, build predefined params and then update with the # user-supplied kwargs, which allows overriding the pre-built # params if needed. params = create_request_parameters(parent, self._waiter_model) params.update(kwargs) logger.debug( 'Calling %s:%s with %r', parent.meta.service_name, self._waiter_resource_name, params, ) client = parent.meta.client waiter = client.get_waiter(client_waiter_name) response = waiter.wait(params) logger.debug('Response: %r', response) class CustomModeledAction: """A custom, modeled action to inject into a resource.""" def __init__(self, action_name, action_model, function, event_emitter): """ :type action_name: str :param action_name: The name of the action to inject, e.g. 'delete_tags' :type action_model: dict :param action_model: A JSON definition of the action, as if it were part of the resource model. :type function: function :param function: The function to perform when the action is called. The first argument should be 'self', which will be the resource the function is to be called on. :type event_emitter: :py:class:`botocore.hooks.BaseEventHooks` :param event_emitter: The session event emitter. """ self.name = action_name self.model = action_model self.function = function self.emitter = event_emitter def inject(self, class_attributes, service_context, event_name, **kwargs): resource_name = event_name.rsplit(".")[-1] action = Action(self.name, self.model, {}) self.function.__name__ = self.name self.function.__doc__ = ActionDocstring( resource_name=resource_name, event_emitter=self.emitter, action_model=action, service_model=service_context.service_model, include_signature=False, ) inject_attribute(class_attributes, self.name, self.function)	9.38 KB	2025-05-25 12:26
📄 base.py # 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__) 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__ 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) 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)))	4.89 KB	2025-05-25 12:26
📄 collection.py # 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 copy import logging from botocore import xform_name from botocore.utils import merge_dicts from ..docs import docstring from .action import BatchAction from .params import create_request_parameters from .response import ResourceHandler logger = logging.getLogger(__name__) class ResourceCollection: """ Represents a collection of resources, which can be iterated through, optionally with filtering. Collections automatically handle pagination for you. See :ref:`guide_collections` for a high-level overview of collections, including when remote service requests are performed. :type model: :py:class:`~boto3.resources.model.Collection` :param model: Collection model :type parent: :py:class:`~boto3.resources.base.ServiceResource` :param parent: The collection's parent resource :type handler: :py:class:`~boto3.resources.response.ResourceHandler` :param handler: The resource response handler used to create resource instances """ def __init__(self, model, parent, handler, kwargs): self._model = model self._parent = parent self._py_operation_name = xform_name(model.request.operation) self._handler = handler self._params = copy.deepcopy(kwargs) def __repr__(self): return '{}({}, {})'.format( self.__class__.__name__, self._parent, f'{self._parent.meta.service_name}.{self._model.resource.type}', ) def __iter__(self): """ A generator which yields resource instances after doing the appropriate service operation calls and handling any pagination on your behalf. Page size, item limit, and filter parameters are applied if they have previously been set. >>> bucket = s3.Bucket('boto3') >>> for obj in bucket.objects.all(): ... print(obj.key) 'key1' 'key2' """ limit = self._params.get('limit', None) count = 0 for page in self.pages(): for item in page: yield item # If the limit is set and has been reached, then # we stop processing items here. count += 1 if limit is not None and count >= limit: return def _clone(self, kwargs): """ Create a clone of this collection. This is used by the methods below to provide a chainable interface that returns copies rather than the original. This allows things like: >>> base = collection.filter(Param1=1) >>> query1 = base.filter(Param2=2) >>> query2 = base.filter(Param3=3) >>> query1.params {'Param1': 1, 'Param2': 2} >>> query2.params {'Param1': 1, 'Param3': 3} :rtype: :py:class:`ResourceCollection` :return: A clone of this resource collection """ params = copy.deepcopy(self._params) merge_dicts(params, kwargs, append_lists=True) clone = self.__class__( self._model, self._parent, self._handler, params ) return clone def pages(self): """ A generator which yields pages of resource instances after doing the appropriate service operation calls and handling any pagination on your behalf. Non-paginated calls will return a single page of items. Page size, item limit, and filter parameters are applied if they have previously been set. >>> bucket = s3.Bucket('boto3') >>> for page in bucket.objects.pages(): ... for obj in page: ... print(obj.key) 'key1' 'key2' :rtype: list(:py:class:`~boto3.resources.base.ServiceResource`) :return: List of resource instances """ client = self._parent.meta.client cleaned_params = self._params.copy() limit = cleaned_params.pop('limit', None) page_size = cleaned_params.pop('page_size', None) params = create_request_parameters(self._parent, self._model.request) merge_dicts(params, cleaned_params, append_lists=True) # Is this a paginated operation? If so, we need to get an # iterator for the various pages. If not, then we simply # call the operation and return the result as a single # page in a list. For non-paginated results, we just ignore # the page size parameter. if client.can_paginate(self._py_operation_name): logger.debug( 'Calling paginated %s:%s with %r', self._parent.meta.service_name, self._py_operation_name, params, ) paginator = client.get_paginator(self._py_operation_name) pages = paginator.paginate( PaginationConfig={'MaxItems': limit, 'PageSize': page_size}, params, ) else: logger.debug( 'Calling %s:%s with %r', self._parent.meta.service_name, self._py_operation_name, params, ) pages = [getattr(client, self._py_operation_name)(params)] # Now that we have a page iterator or single page of results # we start processing and yielding individual items. count = 0 for page in pages: page_items = [] for item in self._handler(self._parent, params, page): page_items.append(item) # If the limit is set and has been reached, then # we stop processing items here. count += 1 if limit is not None and count >= limit: break yield page_items # Stop reading pages if we've reached out limit if limit is not None and count >= limit: break def all(self): """ Get all items from the collection, optionally with a custom page size and item count limit. This method returns an iterable generator which yields individual resource instances. Example use:: # Iterate through items >>> for queue in sqs.queues.all(): ... print(queue.url) 'https://url1' 'https://url2' # Convert to list >>> queues = list(sqs.queues.all()) >>> len(queues) 2 """ return self._clone() def filter(self, kwargs): """ Get items from the collection, passing keyword arguments along as parameters to the underlying service operation, which are typically used to filter the results. This method returns an iterable generator which yields individual resource instances. Example use:: # Iterate through items >>> for queue in sqs.queues.filter(Param='foo'): ... print(queue.url) 'https://url1' 'https://url2' # Convert to list >>> queues = list(sqs.queues.filter(Param='foo')) >>> len(queues) 2 :rtype: :py:class:`ResourceCollection` """ return self._clone(kwargs) def limit(self, count): """ Return at most this many resources. >>> for bucket in s3.buckets.limit(5): ... print(bucket.name) 'bucket1' 'bucket2' 'bucket3' 'bucket4' 'bucket5' :type count: int :param count: Return no more than this many items :rtype: :py:class:`ResourceCollection` """ return self._clone(limit=count) def page_size(self, count): """ Fetch at most this many resources per service request. >>> for obj in s3.Bucket('boto3').objects.page_size(100): ... print(obj.key) :type count: int :param count: Fetch this many items per request :rtype: :py:class:`ResourceCollection` """ return self._clone(page_size=count) class CollectionManager: """ A collection manager provides access to resource collection instances, which can be iterated and filtered. The manager exposes some convenience functions that are also found on resource collections, such as :py:meth:`~ResourceCollection.all` and :py:meth:`~ResourceCollection.filter`. Get all items:: >>> for bucket in s3.buckets.all(): ... print(bucket.name) Get only some items via filtering:: >>> for queue in sqs.queues.filter(QueueNamePrefix='AWS'): ... print(queue.url) Get whole pages of items: >>> for page in s3.Bucket('boto3').objects.pages(): ... for obj in page: ... print(obj.key) A collection manager is not iterable. You must call one of the methods that return a :py:class:`ResourceCollection` before trying to iterate, slice, or convert to a list. See the :ref:`guide_collections` guide for a high-level overview of collections, including when remote service requests are performed. :type collection_model: :py:class:`~boto3.resources.model.Collection` :param model: Collection model :type parent: :py:class:`~boto3.resources.base.ServiceResource` :param parent: The collection's parent resource :type factory: :py:class:`~boto3.resources.factory.ResourceFactory` :param factory: The resource factory to create new resources :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service """ # The class to use when creating an iterator _collection_cls = ResourceCollection def __init__(self, collection_model, parent, factory, service_context): self._model = collection_model operation_name = self._model.request.operation self._parent = parent search_path = collection_model.resource.path self._handler = ResourceHandler( search_path=search_path, factory=factory, resource_model=collection_model.resource, service_context=service_context, operation_name=operation_name, ) def __repr__(self): return '{}({}, {})'.format( self.__class__.__name__, self._parent, f'{self._parent.meta.service_name}.{self._model.resource.type}', ) def iterator(self, kwargs): """ Get a resource collection iterator from this manager. :rtype: :py:class:`ResourceCollection` :return: An iterable representing the collection of resources """ return self._collection_cls( self._model, self._parent, self._handler, kwargs ) # Set up some methods to proxy ResourceCollection methods def all(self): return self.iterator() all.__doc__ = ResourceCollection.all.__doc__ def filter(self, kwargs): return self.iterator(kwargs) filter.__doc__ = ResourceCollection.filter.__doc__ def limit(self, count): return self.iterator(limit=count) limit.__doc__ = ResourceCollection.limit.__doc__ def page_size(self, count): return self.iterator(page_size=count) page_size.__doc__ = ResourceCollection.page_size.__doc__ def pages(self): return self.iterator().pages() pages.__doc__ = ResourceCollection.pages.__doc__ class CollectionFactory: """ A factory to create new :py:class:`CollectionManager` and :py:class:`ResourceCollection` subclasses from a :py:class:`~boto3.resources.model.Collection` model. These subclasses include methods to perform batch operations. """ def load_from_definition( self, resource_name, collection_model, service_context, event_emitter ): """ Loads a collection from a model, creating a new :py:class:`CollectionManager` subclass with the correct properties and methods, named based on the service and resource name, e.g. ec2.InstanceCollectionManager. It also creates a new :py:class:`ResourceCollection` subclass which is used by the new manager class. :type resource_name: string :param resource_name: Name of the resource to look up. For services, this should match the ``service_name``. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service :type event_emitter: :py:class:`~botocore.hooks.HierarchialEmitter` :param event_emitter: An event emitter :rtype: Subclass of :py:class:`CollectionManager` :return: The collection class. """ attrs = {} collection_name = collection_model.name # Create the batch actions for a collection self._load_batch_actions( attrs, resource_name, collection_model, service_context.service_model, event_emitter, ) # Add the documentation to the collection class's methods self._load_documented_collection_methods( attrs=attrs, resource_name=resource_name, collection_model=collection_model, service_model=service_context.service_model, event_emitter=event_emitter, base_class=ResourceCollection, ) if service_context.service_name == resource_name: cls_name = ( f'{service_context.service_name}.{collection_name}Collection' ) else: cls_name = f'{service_context.service_name}.{resource_name}.{collection_name}Collection' collection_cls = type(str(cls_name), (ResourceCollection,), attrs) # Add the documentation to the collection manager's methods self._load_documented_collection_methods( attrs=attrs, resource_name=resource_name, collection_model=collection_model, service_model=service_context.service_model, event_emitter=event_emitter, base_class=CollectionManager, ) attrs['_collection_cls'] = collection_cls cls_name += 'Manager' return type(str(cls_name), (CollectionManager,), attrs) def _load_batch_actions( self, attrs, resource_name, collection_model, service_model, event_emitter, ): """ Batch actions on the collection become methods on both the collection manager and iterators. """ for action_model in collection_model.batch_actions: snake_cased = xform_name(action_model.name) attrs[snake_cased] = self._create_batch_action( resource_name, snake_cased, action_model, collection_model, service_model, event_emitter, ) def _load_documented_collection_methods( factory_self, attrs, resource_name, collection_model, service_model, event_emitter, base_class, ): # The base class already has these methods defined. However # the docstrings are generic and not based for a particular service # or resource. So we override these methods by proxying to the # base class's builtin method and adding a docstring # that pertains to the resource. # A collection's all() method. def all(self): return base_class.all(self) all.__doc__ = docstring.CollectionMethodDocstring( resource_name=resource_name, action_name='all', event_emitter=event_emitter, collection_model=collection_model, service_model=service_model, include_signature=False, ) attrs['all'] = all # The collection's filter() method. def filter(self, kwargs): return base_class.filter(self, *kwargs) filter.__doc__ = docstring.CollectionMethodDocstring( resource_name=resource_name, action_name='filter', event_emitter=event_emitter, collection_model=collection_model, service_model=service_model, include_signature=False, ) attrs['filter'] = filter # The collection's limit method. def limit(self, count): return base_class.limit(self, count) limit.__doc__ = docstring.CollectionMethodDocstring( resource_name=resource_name, action_name='limit', event_emitter=event_emitter, collection_model=collection_model, service_model=service_model, include_signature=False, ) attrs['limit'] = limit # The collection's page_size method. def page_size(self, count): return base_class.page_size(self, count) page_size.__doc__ = docstring.CollectionMethodDocstring( resource_name=resource_name, action_name='page_size', event_emitter=event_emitter, collection_model=collection_model, service_model=service_model, include_signature=False, ) attrs['page_size'] = page_size def _create_batch_action( factory_self, resource_name, snake_cased, action_model, collection_model, service_model, event_emitter, ): """ Creates a new method which makes a batch operation request to the underlying service API. """ action = BatchAction(action_model) def batch_action(self, args, *kwargs): return action(self, args, **kwargs) batch_action.__name__ = str(snake_cased) batch_action.__doc__ = docstring.BatchActionDocstring( resource_name=resource_name, event_emitter=event_emitter, batch_action_model=action_model, service_model=service_model, collection_model=collection_model, include_signature=False, ) return batch_action	18.67 KB	2025-05-25 12:26
📄 factory.py # 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 from functools import partial from ..docs import docstring from ..exceptions import ResourceLoadException from .action import ServiceAction, WaiterAction from .base import ResourceMeta, ServiceResource from .collection import CollectionFactory from .model import ResourceModel from .response import ResourceHandler, build_identifiers logger = logging.getLogger(__name__) class ResourceFactory: """ A factory to create new :py:class:`~boto3.resources.base.ServiceResource` classes from a :py:class:`~boto3.resources.model.ResourceModel`. There are two types of lookups that can be done: one on the service itself (e.g. an SQS resource) and another on models contained within the service (e.g. an SQS Queue resource). """ def __init__(self, emitter): self._collection_factory = CollectionFactory() self._emitter = emitter def load_from_definition( self, resource_name, single_resource_json_definition, service_context ): """ Loads a resource from a model, creating a new :py:class:`~boto3.resources.base.ServiceResource` subclass with the correct properties and methods, named based on the service and resource name, e.g. EC2.Instance. :type resource_name: string :param resource_name: Name of the resource to look up. For services, this should match the ``service_name``. :type single_resource_json_definition: dict :param single_resource_json_definition: The loaded json of a single service resource or resource definition. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service :rtype: Subclass of :py:class:`~boto3.resources.base.ServiceResource` :return: The service or resource class. """ logger.debug( 'Loading %s:%s', service_context.service_name, resource_name ) # Using the loaded JSON create a ResourceModel object. resource_model = ResourceModel( resource_name, single_resource_json_definition, service_context.resource_json_definitions, ) # Do some renaming of the shape if there was a naming collision # that needed to be accounted for. shape = None if resource_model.shape: shape = service_context.service_model.shape_for( resource_model.shape ) resource_model.load_rename_map(shape) # Set some basic info meta = ResourceMeta( service_context.service_name, resource_model=resource_model ) attrs = { 'meta': meta, } # Create and load all of attributes of the resource class based # on the models. # Identifiers self._load_identifiers( attrs=attrs, meta=meta, resource_name=resource_name, resource_model=resource_model, ) # Load/Reload actions self._load_actions( attrs=attrs, resource_name=resource_name, resource_model=resource_model, service_context=service_context, ) # Attributes that get auto-loaded self._load_attributes( attrs=attrs, meta=meta, resource_name=resource_name, resource_model=resource_model, service_context=service_context, ) # Collections and their corresponding methods self._load_collections( attrs=attrs, resource_model=resource_model, service_context=service_context, ) # References and Subresources self._load_has_relations( attrs=attrs, resource_name=resource_name, resource_model=resource_model, service_context=service_context, ) # Waiter resource actions self._load_waiters( attrs=attrs, resource_name=resource_name, resource_model=resource_model, service_context=service_context, ) # Create the name based on the requested service and resource cls_name = resource_name if service_context.service_name == resource_name: cls_name = 'ServiceResource' cls_name = service_context.service_name + '.' + cls_name base_classes = [ServiceResource] if self._emitter is not None: self._emitter.emit( f'creating-resource-class.{cls_name}', class_attributes=attrs, base_classes=base_classes, service_context=service_context, ) return type(str(cls_name), tuple(base_classes), attrs) def _load_identifiers(self, attrs, meta, resource_model, resource_name): """ Populate required identifiers. These are arguments without which the resource cannot be used. Identifiers become arguments for operations on the resource. """ for identifier in resource_model.identifiers: meta.identifiers.append(identifier.name) attrs[identifier.name] = self._create_identifier( identifier, resource_name ) def _load_actions( self, attrs, resource_name, resource_model, service_context ): """ Actions on the resource become methods, with the ``load`` method being a special case which sets internal data for attributes, and ``reload`` is an alias for ``load``. """ if resource_model.load: attrs['load'] = self._create_action( action_model=resource_model.load, resource_name=resource_name, service_context=service_context, is_load=True, ) attrs['reload'] = attrs['load'] for action in resource_model.actions: attrs[action.name] = self._create_action( action_model=action, resource_name=resource_name, service_context=service_context, ) def _load_attributes( self, attrs, meta, resource_name, resource_model, service_context ): """ Load resource attributes based on the resource shape. The shape name is referenced in the resource JSON, but the shape itself is defined in the Botocore service JSON, hence the need for access to the ``service_model``. """ if not resource_model.shape: return shape = service_context.service_model.shape_for(resource_model.shape) identifiers = { i.member_name: i for i in resource_model.identifiers if i.member_name } attributes = resource_model.get_attributes(shape) for name, (orig_name, member) in attributes.items(): if name in identifiers: prop = self._create_identifier_alias( resource_name=resource_name, identifier=identifiers[name], member_model=member, service_context=service_context, ) else: prop = self._create_autoload_property( resource_name=resource_name, name=orig_name, snake_cased=name, member_model=member, service_context=service_context, ) attrs[name] = prop def _load_collections(self, attrs, resource_model, service_context): """ Load resource collections from the model. Each collection becomes a :py:class:`~boto3.resources.collection.CollectionManager` instance on the resource instance, which allows you to iterate and filter through the collection's items. """ for collection_model in resource_model.collections: attrs[collection_model.name] = self._create_collection( resource_name=resource_model.name, collection_model=collection_model, service_context=service_context, ) def _load_has_relations( self, attrs, resource_name, resource_model, service_context ): """ Load related resources, which are defined via a ``has`` relationship but conceptually come in two forms: 1. A reference, which is a related resource instance and can be ``None``, such as an EC2 instance's ``vpc``. 2. A subresource, which is a resource constructor that will always return a resource instance which shares identifiers/data with this resource, such as ``s3.Bucket('name').Object('key')``. """ for reference in resource_model.references: # This is a dangling reference, i.e. we have all # the data we need to create the resource, so # this instance becomes an attribute on the class. attrs[reference.name] = self._create_reference( reference_model=reference, resource_name=resource_name, service_context=service_context, ) for subresource in resource_model.subresources: # This is a sub-resource class you can create # by passing in an identifier, e.g. s3.Bucket(name). attrs[subresource.name] = self._create_class_partial( subresource_model=subresource, resource_name=resource_name, service_context=service_context, ) self._create_available_subresources_command( attrs, resource_model.subresources ) def _create_available_subresources_command(self, attrs, subresources): _subresources = [subresource.name for subresource in subresources] _subresources = sorted(_subresources) def get_available_subresources(factory_self): """ Returns a list of all the available sub-resources for this Resource. :returns: A list containing the name of each sub-resource for this resource :rtype: list of str """ return _subresources attrs['get_available_subresources'] = get_available_subresources def _load_waiters( self, attrs, resource_name, resource_model, service_context ): """ Load resource waiters from the model. Each waiter allows you to wait until a resource reaches a specific state by polling the state of the resource. """ for waiter in resource_model.waiters: attrs[waiter.name] = self._create_waiter( resource_waiter_model=waiter, resource_name=resource_name, service_context=service_context, ) def _create_identifier(factory_self, identifier, resource_name): """ Creates a read-only property for identifier attributes. """ def get_identifier(self): # The default value is set to ``None`` instead of # raising an AttributeError because when resources are # instantiated a check is made such that none of the # identifiers have a value ``None``. If any are ``None``, # a more informative user error than a generic AttributeError # is raised. return getattr(self, '_' + identifier.name, None) get_identifier.__name__ = str(identifier.name) get_identifier.__doc__ = docstring.IdentifierDocstring( resource_name=resource_name, identifier_model=identifier, include_signature=False, ) return property(get_identifier) def _create_identifier_alias( factory_self, resource_name, identifier, member_model, service_context ): """ Creates a read-only property that aliases an identifier. """ def get_identifier(self): return getattr(self, '_' + identifier.name, None) get_identifier.__name__ = str(identifier.member_name) get_identifier.__doc__ = docstring.AttributeDocstring( service_name=service_context.service_name, resource_name=resource_name, attr_name=identifier.member_name, event_emitter=factory_self._emitter, attr_model=member_model, include_signature=False, ) return property(get_identifier) def _create_autoload_property( factory_self, resource_name, name, snake_cased, member_model, service_context, ): """ Creates a new property on the resource to lazy-load its value via the resource's ``load`` method (if it exists). """ # The property loader will check to see if this resource has already # been loaded and return the cached value if possible. If not, then # it first checks to see if it CAN be loaded (raise if not), then # calls the load before returning the value. def property_loader(self): if self.meta.data is None: if hasattr(self, 'load'): self.load() else: raise ResourceLoadException( f'{self.__class__.__name__} has no load method' ) return self.meta.data.get(name) property_loader.__name__ = str(snake_cased) property_loader.__doc__ = docstring.AttributeDocstring( service_name=service_context.service_name, resource_name=resource_name, attr_name=snake_cased, event_emitter=factory_self._emitter, attr_model=member_model, include_signature=False, ) return property(property_loader) def _create_waiter( factory_self, resource_waiter_model, resource_name, service_context ): """ Creates a new wait method for each resource where both a waiter and resource model is defined. """ waiter = WaiterAction( resource_waiter_model, waiter_resource_name=resource_waiter_model.name, ) def do_waiter(self, args, kwargs): waiter(self, args, *kwargs) do_waiter.__name__ = str(resource_waiter_model.name) do_waiter.__doc__ = docstring.ResourceWaiterDocstring( resource_name=resource_name, event_emitter=factory_self._emitter, service_model=service_context.service_model, resource_waiter_model=resource_waiter_model, service_waiter_model=service_context.service_waiter_model, include_signature=False, ) return do_waiter def _create_collection( factory_self, resource_name, collection_model, service_context ): """ Creates a new property on the resource to lazy-load a collection. """ cls = factory_self._collection_factory.load_from_definition( resource_name=resource_name, collection_model=collection_model, service_context=service_context, event_emitter=factory_self._emitter, ) def get_collection(self): return cls( collection_model=collection_model, parent=self, factory=factory_self, service_context=service_context, ) get_collection.__name__ = str(collection_model.name) get_collection.__doc__ = docstring.CollectionDocstring( collection_model=collection_model, include_signature=False ) return property(get_collection) def _create_reference( factory_self, reference_model, resource_name, service_context ): """ Creates a new property on the resource to lazy-load a reference. """ # References are essentially an action with no request # or response, so we can re-use the response handlers to # build up resources from identifiers and data members. handler = ResourceHandler( search_path=reference_model.resource.path, factory=factory_self, resource_model=reference_model.resource, service_context=service_context, ) # Are there any identifiers that need access to data members? # This is important when building the resource below since # it requires the data to be loaded. needs_data = any( i.source == 'data' for i in reference_model.resource.identifiers ) def get_reference(self): # We need to lazy-evaluate the reference to handle circular # references between resources. We do this by loading the class # when first accessed. # This is using a response handler* so we need to make sure # our data is loaded (if possible) and pass that data into # the handler as if it were a response. This allows references # to have their data loaded properly. if needs_data and self.meta.data is None and hasattr(self, 'load'): self.load() return handler(self, {}, self.meta.data) get_reference.__name__ = str(reference_model.name) get_reference.__doc__ = docstring.ReferenceDocstring( reference_model=reference_model, include_signature=False ) return property(get_reference) def _create_class_partial( factory_self, subresource_model, resource_name, service_context ): """ Creates a new method which acts as a functools.partial, passing along the instance's low-level `client` to the new resource class' constructor. """ name = subresource_model.resource.type def create_resource(self, args, kwargs): # We need a new method here because we want access to the # instance's client. positional_args = [] # We lazy-load the class to handle circular references. json_def = service_context.resource_json_definitions.get(name, {}) resource_cls = factory_self.load_from_definition( resource_name=name, single_resource_json_definition=json_def, service_context=service_context, ) # Assumes that identifiers are in order, which lets you do # e.g. ``sqs.Queue('foo').Message('bar')`` to create a new message # linked with the ``foo`` queue and which has a ``bar`` receipt # handle. If we did kwargs here then future positional arguments # would lead to failure. identifiers = subresource_model.resource.identifiers if identifiers is not None: for identifier, value in build_identifiers(identifiers, self): positional_args.append(value) return partial( resource_cls, positional_args, client=self.meta.client )(args, kwargs) create_resource.__name__ = str(name) create_resource.__doc__ = docstring.SubResourceDocstring( resource_name=resource_name, sub_resource_model=subresource_model, service_model=service_context.service_model, include_signature=False, ) return create_resource def _create_action( factory_self, action_model, resource_name, service_context, is_load=False, ): """ Creates a new method which makes a request to the underlying AWS service. """ # Create the action in in this closure but before the ``do_action`` # method below is invoked, which allows instances of the resource # to share the ServiceAction instance. action = ServiceAction( action_model, factory=factory_self, service_context=service_context ) # A resource's ``load`` method is special because it sets # values on the resource instead of returning the response. if is_load: # We need a new method here because we want access to the # instance via ``self``. def do_action(self, args, *kwargs): response = action(self, args, *kwargs) self.meta.data = response # Create the docstring for the load/reload methods. lazy_docstring = docstring.LoadReloadDocstring( action_name=action_model.name, resource_name=resource_name, event_emitter=factory_self._emitter, load_model=action_model, service_model=service_context.service_model, include_signature=False, ) else: # We need a new method here because we want access to the # instance via ``self``. def do_action(self, args, *kwargs): response = action(self, args, **kwargs) if hasattr(self, 'load'): # Clear cached data. It will be reloaded the next # time that an attribute is accessed. # TODO: Make this configurable in the future? self.meta.data = None return response lazy_docstring = docstring.ActionDocstring( resource_name=resource_name, event_emitter=factory_self._emitter, action_model=action_model, service_model=service_context.service_model, include_signature=False, ) do_action.__name__ = str(action_model.name) do_action.__doc__ = lazy_docstring return do_action	22.18 KB	2025-05-25 12:26
📄 model.py # 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__) 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 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') 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 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) 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') 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') 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 ) 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 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') 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) 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	19.86 KB	2025-05-25 12:26
📄 params.py # 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 re import jmespath from botocore import xform_name from ..exceptions import ResourceLoadException INDEX_RE = re.compile(r'\[(.)\]$') def get_data_member(parent, path): """ Get a data member from a parent using a JMESPath search query, loading the parent if required. If the parent cannot be loaded and no data is present then an exception is raised. :type parent: ServiceResource :param parent: The resource instance to which contains data we are interested in. :type path: string :param path: The JMESPath expression to query :raises ResourceLoadException: When no data is present and the resource cannot be loaded. :returns: The queried data or ``None``. """ # Ensure the parent has its data loaded, if possible. if parent.meta.data is None: if hasattr(parent, 'load'): parent.load() else: raise ResourceLoadException( f'{parent.__class__.__name__} has no load method!' ) return jmespath.search(path, parent.meta.data) def create_request_parameters(parent, request_model, params=None, index=None): """ Handle request parameters that can be filled in from identifiers, resource data members or constants. By passing ``params``, you can invoke this method multiple times and build up a parameter dict over time, which is particularly useful for reverse JMESPath expressions that append to lists. :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type request_model: :py:class:`~boto3.resources.model.Request` :param request_model: The action request model. :type params: dict :param params: If set, then add to this existing dict. It is both edited in-place and returned. :type index: int :param index: The position of an item within a list :rtype: dict :return: Pre-filled parameters to be sent to the request operation. """ if params is None: params = {} for param in request_model.params: source = param.source target = param.target if source == 'identifier': # Resource identifier, e.g. queue.url value = getattr(parent, xform_name(param.name)) elif source == 'data': # If this is a data member then it may incur a load # action before returning the value. value = get_data_member(parent, param.path) elif source in ['string', 'integer', 'boolean']: # These are hard-coded values in the definition value = param.value elif source == 'input': # This is provided by the user, so ignore it here continue else: raise NotImplementedError(f'Unsupported source type: {source}') build_param_structure(params, target, value, index) return params def build_param_structure(params, target, value, index=None): """ This method provides a basic reverse JMESPath implementation that lets you go from a JMESPath-like string to a possibly deeply nested object. The ``params`` are mutated in-place, so subsequent calls can modify the same element by its index. >>> build_param_structure(params, 'test[0]', 1) >>> print(params) {'test': [1]} >>> build_param_structure(params, 'foo.bar[0].baz', 'hello world') >>> print(params) {'test': [1], 'foo': {'bar': [{'baz': 'hello, world'}]}} """ pos = params parts = target.split('.') # First, split into parts like 'foo', 'bar[0]', 'baz' and process # each piece. It can either be a list or a dict, depending on if # an index like `[0]` is present. We detect this via a regular # expression, and keep track of where we are in params via the # pos variable, walking down to the last item. Once there, we # set the value. for i, part in enumerate(parts): # Is it indexing an array? result = INDEX_RE.search(part) if result: if result.group(1): if result.group(1) == '': part = part[:-3] else: # We have an explicit index index = int(result.group(1)) part = part[: -len(str(index) + '[]')] else: # Index will be set after we know the proper part # name and that it's a list instance. index = None part = part[:-2] if part not in pos or not isinstance(pos[part], list): pos[part] = [] # This means we should append, e.g. 'foo[]' if index is None: index = len(pos[part]) while len(pos[part]) <= index: # Assume it's a dict until we set the final value below pos[part].append({}) # Last item? Set the value, otherwise set the new position if i == len(parts) - 1: pos[part][index] = value else: # The new pos is the item in the array, not the array! pos = pos[part][index] else: if part not in pos: pos[part] = {} # Last item? Set the value, otherwise set the new position if i == len(parts) - 1: pos[part] = value else: pos = pos[part]	5.97 KB	2025-05-25 12:26
📄 response.py # 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 jmespath from botocore import xform_name from .params import get_data_member def all_not_none(iterable): """ Return True if all elements of the iterable are not None (or if the iterable is empty). This is like the built-in ``all``, except checks against None, so 0 and False are allowable values. """ for element in iterable: if element is None: return False return True def build_identifiers(identifiers, parent, params=None, raw_response=None): """ Builds a mapping of identifier names to values based on the identifier source location, type, and target. Identifier values may be scalars or lists depending on the source type and location. :type identifiers: list :param identifiers: List of :py:class:`~boto3.resources.model.Parameter` definitions :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type raw_response: dict :param raw_response: Low-level operation response. :rtype: list :return: An ordered list of ``(name, value)`` identifier tuples. """ results = [] for identifier in identifiers: source = identifier.source target = identifier.target if source == 'response': value = jmespath.search(identifier.path, raw_response) elif source == 'requestParameter': value = jmespath.search(identifier.path, params) elif source == 'identifier': value = getattr(parent, xform_name(identifier.name)) elif source == 'data': # If this is a data member then it may incur a load # action before returning the value. value = get_data_member(parent, identifier.path) elif source == 'input': # This value is set by the user, so ignore it here continue else: raise NotImplementedError(f'Unsupported source type: {source}') results.append((xform_name(target), value)) return results def build_empty_response(search_path, operation_name, service_model): """ Creates an appropriate empty response for the type that is expected, based on the service model's shape type. For example, a value that is normally a list would then return an empty list. A structure would return an empty dict, and a number would return None. :type search_path: string :param search_path: JMESPath expression to search in the response :type operation_name: string :param operation_name: Name of the underlying service operation. :type service_model: :ref:`botocore.model.ServiceModel` :param service_model: The Botocore service model :rtype: dict, list, or None :return: An appropriate empty value """ response = None operation_model = service_model.operation_model(operation_name) shape = operation_model.output_shape if search_path: # Walk the search path and find the final shape. For example, given # a path of ``foo.bar[0].baz``, we first find the shape for ``foo``, # then the shape for ``bar`` (ignoring the indexing), and finally # the shape for ``baz``. for item in search_path.split('.'): item = item.strip('[0123456789]$') if shape.type_name == 'structure': shape = shape.members[item] elif shape.type_name == 'list': shape = shape.member else: raise NotImplementedError( f'Search path hits shape type {shape.type_name} from {item}' ) # Anything not handled here is set to None if shape.type_name == 'structure': response = {} elif shape.type_name == 'list': response = [] elif shape.type_name == 'map': response = {} return response class RawHandler: """ A raw action response handler. This passed through the response dictionary, optionally after performing a JMESPath search if one has been defined for the action. :type search_path: string :param search_path: JMESPath expression to search in the response :rtype: dict :return: Service response """ def __init__(self, search_path): self.search_path = search_path def __call__(self, parent, params, response): """ :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type response: dict :param response: Low-level operation response. """ # TODO: Remove the '$' check after JMESPath supports it if self.search_path and self.search_path != '$': response = jmespath.search(self.search_path, response) return response class ResourceHandler: """ Creates a new resource or list of new resources from the low-level response based on the given response resource definition. :type search_path: string :param search_path: JMESPath expression to search in the response :type factory: ResourceFactory :param factory: The factory that created the resource class to which this action is attached. :type resource_model: :py:class:`~boto3.resources.model.ResponseResource` :param resource_model: Response resource model. :type service_context: :py:class:`~boto3.utils.ServiceContext` :param service_context: Context about the AWS service :type operation_name: string :param operation_name: Name of the underlying service operation, if it exists. :rtype: ServiceResource or list :return: New resource instance(s). """ def __init__( self, search_path, factory, resource_model, service_context, operation_name=None, ): self.search_path = search_path self.factory = factory self.resource_model = resource_model self.operation_name = operation_name self.service_context = service_context def __call__(self, parent, params, response): """ :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type params: dict :param params: Request parameters sent to the service. :type response: dict :param response: Low-level operation response. """ resource_name = self.resource_model.type json_definition = self.service_context.resource_json_definitions.get( resource_name ) # Load the new resource class that will result from this action. resource_cls = self.factory.load_from_definition( resource_name=resource_name, single_resource_json_definition=json_definition, service_context=self.service_context, ) raw_response = response search_response = None # Anytime a path is defined, it means the response contains the # resource's attributes, so resource_data gets set here. It # eventually ends up in resource.meta.data, which is where # the attribute properties look for data. if self.search_path: search_response = jmespath.search(self.search_path, raw_response) # First, we parse all the identifiers, then create the individual # response resources using them. Any identifiers that are lists # will have one item consumed from the front of the list for each # resource that is instantiated. Items which are not a list will # be set as the same value on each new resource instance. identifiers = dict( build_identifiers( self.resource_model.identifiers, parent, params, raw_response ) ) # If any of the identifiers is a list, then the response is plural plural = [v for v in identifiers.values() if isinstance(v, list)] if plural: response = [] # The number of items in an identifier that is a list will # determine how many resource instances to create. for i in range(len(plural[0])): # Response item data is only available if a search path # was given. This prevents accidentally loading unrelated # data that may be in the response. response_item = None if search_response: response_item = search_response[i] response.append( self.handle_response_item( resource_cls, parent, identifiers, response_item ) ) elif all_not_none(identifiers.values()): # All identifiers must always exist, otherwise the resource # cannot be instantiated. response = self.handle_response_item( resource_cls, parent, identifiers, search_response ) else: # The response should be empty, but that may mean an # empty dict, list, or None based on whether we make # a remote service call and what shape it is expected # to return. response = None if self.operation_name is not None: # A remote service call was made, so try and determine # its shape. response = build_empty_response( self.search_path, self.operation_name, self.service_context.service_model, ) return response def handle_response_item( self, resource_cls, parent, identifiers, resource_data ): """ Handles the creation of a single response item by setting parameters and creating the appropriate resource instance. :type resource_cls: ServiceResource subclass :param resource_cls: The resource class to instantiate. :type parent: ServiceResource :param parent: The resource instance to which this action is attached. :type identifiers: dict :param identifiers: Map of identifier names to value or values. :type resource_data: dict or None :param resource_data: Data for resource attributes. :rtype: ServiceResource :return: New resource instance. """ kwargs = { 'client': parent.meta.client, } for name, value in identifiers.items(): # If value is a list, then consume the next item if isinstance(value, list): value = value.pop(0) kwargs[name] = value resource = resource_cls(**kwargs) if resource_data is not None: resource.meta.data = resource_data return resource	11.37 KB	2025-05-25 12:26

Terminal