diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/boto3/resources')
8 files changed, 2690 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/__init__.py b/.venv/lib/python3.12/site-packages/boto3/resources/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/__init__.py diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/action.py b/.venv/lib/python3.12/site-packages/boto3/resources/action.py new file mode 100644 index 00000000..7c7d8392 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/action.py @@ -0,0 +1,257 @@ +# 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) diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/base.py b/.venv/lib/python3.12/site-packages/boto3/resources/base.py new file mode 100644 index 00000000..78fa9199 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/base.py @@ -0,0 +1,153 @@ +# 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))) diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/collection.py b/.venv/lib/python3.12/site-packages/boto3/resources/collection.py new file mode 100644 index 00000000..5d4c9e9d --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/collection.py @@ -0,0 +1,566 @@ +# 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 diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/factory.py b/.venv/lib/python3.12/site-packages/boto3/resources/factory.py new file mode 100644 index 00000000..4cdd2f01 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/factory.py @@ -0,0 +1,601 @@ +# 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 diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/model.py b/.venv/lib/python3.12/site-packages/boto3/resources/model.py new file mode 100644 index 00000000..9d61fea9 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/model.py @@ -0,0 +1,630 @@ +# 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 diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/params.py b/.venv/lib/python3.12/site-packages/boto3/resources/params.py new file mode 100644 index 00000000..3c5c74b3 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/params.py @@ -0,0 +1,167 @@ +# 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] diff --git a/.venv/lib/python3.12/site-packages/boto3/resources/response.py b/.venv/lib/python3.12/site-packages/boto3/resources/response.py new file mode 100644 index 00000000..a27190a0 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/boto3/resources/response.py @@ -0,0 +1,316 @@ +# 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 |