Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use resource_name method in localstack

Best Python code snippet using localstack_python

factory.py

Source:factory.py

1# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# https://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import logging14from functools import partial15from .action import ServiceAction16from .action import WaiterAction17from .base import ResourceMeta, ServiceResource18from .collection import CollectionFactory19from .model import ResourceModel20from .response import build_identifiers, ResourceHandler21from ..exceptions import ResourceLoadException22from ..docs import docstring23logger = logging.getLogger(__name__)24class ResourceFactory(object):25    """26    A factory to create new :py:class:`~boto3.resources.base.ServiceResource`27    classes from a :py:class:`~boto3.resources.model.ResourceModel`. There are28    two types of lookups that can be done: one on the service itself (e.g. an29    SQS resource) and another on models contained within the service (e.g. an30    SQS Queue resource).31    """32    def __init__(self, emitter):33        self._collection_factory = CollectionFactory()34        self._emitter = emitter35    def load_from_definition(self, resource_name,36                             single_resource_json_definition, service_context):37        """38        Loads a resource from a model, creating a new39        :py:class:`~boto3.resources.base.ServiceResource` subclass40        with the correct properties and methods, named based on the service41        and resource name, e.g. EC2.Instance.42        :type resource_name: string43        :param resource_name: Name of the resource to look up. For services,44                              this should match the ``service_name``.45        :type single_resource_json_definition: dict46        :param single_resource_json_definition:47            The loaded json of a single service resource or resource48            definition.49        :type service_context: :py:class:`~boto3.utils.ServiceContext`50        :param service_context: Context about the AWS service51        :rtype: Subclass of :py:class:`~boto3.resources.base.ServiceResource`52        :return: The service or resource class.53        """54        logger.debug('Loading %s:%s', service_context.service_name,55                     resource_name)56        # Using the loaded JSON create a ResourceModel object.57        resource_model = ResourceModel(58            resource_name, single_resource_json_definition,59            service_context.resource_json_definitions60        )61        # Do some renaming of the shape if there was a naming collision62        # that needed to be accounted for.63        shape = None64        if resource_model.shape:65            shape = service_context.service_model.shape_for(66                resource_model.shape)67        resource_model.load_rename_map(shape)68        # Set some basic info69        meta = ResourceMeta(70            service_context.service_name, resource_model=resource_model)71        attrs = {72            'meta': meta,73        }74        # Create and load all of attributes of the resource class based75        # on the models.76        # Identifiers77        self._load_identifiers(78            attrs=attrs, meta=meta, resource_name=resource_name,79            resource_model=resource_model80        )81        # Load/Reload actions82        self._load_actions(83            attrs=attrs, resource_name=resource_name,84            resource_model=resource_model, service_context=service_context85        )86        # Attributes that get auto-loaded87        self._load_attributes(88            attrs=attrs, meta=meta, resource_name=resource_name,89            resource_model=resource_model,90            service_context=service_context)91        # Collections and their corresponding methods92        self._load_collections(93            attrs=attrs, resource_model=resource_model,94            service_context=service_context)95        # References and Subresources96        self._load_has_relations(97            attrs=attrs, resource_name=resource_name,98            resource_model=resource_model, service_context=service_context99        )100        # Waiter resource actions101        self._load_waiters(102            attrs=attrs, resource_name=resource_name,103            resource_model=resource_model, service_context=service_context104        )105        # Create the name based on the requested service and resource106        cls_name = resource_name107        if service_context.service_name == resource_name:108            cls_name = 'ServiceResource'109        cls_name = service_context.service_name + '.' + cls_name110        base_classes = [ServiceResource]111        if self._emitter is not None:112            self._emitter.emit(113                'creating-resource-class.%s' % cls_name,114                class_attributes=attrs, base_classes=base_classes,115                service_context=service_context)116        return type(str(cls_name), tuple(base_classes), attrs)117    def _load_identifiers(self, attrs, meta, resource_model, resource_name):118        """119        Populate required identifiers. These are arguments without which120        the resource cannot be used. Identifiers become arguments for121        operations on the resource.122        """123        for identifier in resource_model.identifiers:124            meta.identifiers.append(identifier.name)125            attrs[identifier.name] = self._create_identifier(126                identifier, resource_name)127    def _load_actions(self, attrs, resource_name, resource_model,128                      service_context):129        """130        Actions on the resource become methods, with the ``load`` method131        being a special case which sets internal data for attributes, and132        ``reload`` is an alias for ``load``.133        """134        if resource_model.load:135            attrs['load'] = self._create_action(136                action_model=resource_model.load, resource_name=resource_name,137                service_context=service_context, is_load=True)138            attrs['reload'] = attrs['load']139        for action in resource_model.actions:140            attrs[action.name] = self._create_action(141                action_model=action, resource_name=resource_name,142                service_context=service_context)143    def _load_attributes(self, attrs, meta, resource_name, resource_model,144                         service_context):145        """146        Load resource attributes based on the resource shape. The shape147        name is referenced in the resource JSON, but the shape itself148        is defined in the Botocore service JSON, hence the need for149        access to the ``service_model``.150        """151        if not resource_model.shape:152            return153        shape = service_context.service_model.shape_for(154            resource_model.shape)155        identifiers = dict(156            (i.member_name, i)157            for i in resource_model.identifiers if i.member_name)158        attributes = resource_model.get_attributes(shape)159        for name, (orig_name, member) in attributes.items():160            if name in identifiers:161                prop = self._create_identifier_alias(162                    resource_name=resource_name,163                    identifier=identifiers[name],164                    member_model=member,165                    service_context=service_context166                )167            else:168                prop = self._create_autoload_property(169                    resource_name=resource_name,170                    name=orig_name, snake_cased=name,171                    member_model=member,172                    service_context=service_context173                )174            attrs[name] = prop175    def _load_collections(self, attrs, resource_model, service_context):176        """177        Load resource collections from the model. Each collection becomes178        a :py:class:`~boto3.resources.collection.CollectionManager` instance179        on the resource instance, which allows you to iterate and filter180        through the collection's items.181        """182        for collection_model in resource_model.collections:183            attrs[collection_model.name] = self._create_collection(184                resource_name=resource_model.name,185                collection_model=collection_model,186                service_context=service_context187            )188    def _load_has_relations(self, attrs, resource_name, resource_model,189                            service_context):190        """191        Load related resources, which are defined via a ``has``192        relationship but conceptually come in two forms:193        1. A reference, which is a related resource instance and can be194           ``None``, such as an EC2 instance's ``vpc``.195        2. A subresource, which is a resource constructor that will always196           return a resource instance which shares identifiers/data with197           this resource, such as ``s3.Bucket('name').Object('key')``.198        """199        for reference in resource_model.references:200            # This is a dangling reference, i.e. we have all201            # the data we need to create the resource, so202            # this instance becomes an attribute on the class.203            attrs[reference.name] = self._create_reference(204                reference_model=reference,205                resource_name=resource_name,206                service_context=service_context207            )208        for subresource in resource_model.subresources:209            # This is a sub-resource class you can create210            # by passing in an identifier, e.g. s3.Bucket(name).211            attrs[subresource.name] = self._create_class_partial(212                subresource_model=subresource,213                resource_name=resource_name,214                service_context=service_context215            )216        self._create_available_subresources_command(217            attrs, resource_model.subresources)218    def _create_available_subresources_command(self, attrs, subresources):219        _subresources = [subresource.name for subresource in subresources]220        _subresources = sorted(_subresources)221        def get_available_subresources(factory_self):222            """223            Returns a list of all the available sub-resources for this224            Resource.225            :returns: A list containing the name of each sub-resource for this226                resource227            :rtype: list of str228            """229            return _subresources230        attrs['get_available_subresources'] = get_available_subresources231    def _load_waiters(self, attrs, resource_name, resource_model,232                      service_context):233        """234        Load resource waiters from the model. Each waiter allows you to235        wait until a resource reaches a specific state by polling the state236        of the resource.237        """238        for waiter in resource_model.waiters:239            attrs[waiter.name] = self._create_waiter(240                resource_waiter_model=waiter,241                resource_name=resource_name,242                service_context=service_context243            )244    def _create_identifier(factory_self, identifier, resource_name):245        """246        Creates a read-only property for identifier attributes.247        """248        def get_identifier(self):249            # The default value is set to ``None`` instead of250            # raising an AttributeError because when resources are251            # instantiated a check is made such that none of the252            # identifiers have a value ``None``. If any are ``None``,253            # a more informative user error than a generic AttributeError254            # is raised.255            return getattr(self, '_' + identifier.name, None)256        get_identifier.__name__ = str(identifier.name)257        get_identifier.__doc__ = docstring.IdentifierDocstring(258            resource_name=resource_name,259            identifier_model=identifier,260            include_signature=False261        )262        return property(get_identifier)263    def _create_identifier_alias(factory_self, resource_name, identifier,264                                 member_model, service_context):265        """266        Creates a read-only property that aliases an identifier.267        """268        def get_identifier(self):269            return getattr(self, '_' + identifier.name, None)270        get_identifier.__name__ = str(identifier.member_name)271        get_identifier.__doc__ = docstring.AttributeDocstring(272            service_name=service_context.service_name,273            resource_name=resource_name,274            attr_name=identifier.member_name,275            event_emitter=factory_self._emitter,276            attr_model=member_model,277            include_signature=False278        )279        return property(get_identifier)280    def _create_autoload_property(factory_self, resource_name, name,281                                  snake_cased, member_model, service_context):282        """283        Creates a new property on the resource to lazy-load its value284        via the resource's ``load`` method (if it exists).285        """286        # The property loader will check to see if this resource has already287        # been loaded and return the cached value if possible. If not, then288        # it first checks to see if it CAN be loaded (raise if not), then289        # calls the load before returning the value.290        def property_loader(self):291            if self.meta.data is None:292                if hasattr(self, 'load'):293                    self.load()294                else:295                    raise ResourceLoadException(296                        '{0} has no load method'.format(297                            self.__class__.__name__))298            return self.meta.data.get(name)299        property_loader.__name__ = str(snake_cased)300        property_loader.__doc__ = docstring.AttributeDocstring(301            service_name=service_context.service_name,302            resource_name=resource_name,303            attr_name=snake_cased,304            event_emitter=factory_self._emitter,305            attr_model=member_model,306            include_signature=False307        )308        return property(property_loader)309    def _create_waiter(factory_self, resource_waiter_model, resource_name,310                       service_context):311        """312        Creates a new wait method for each resource where both a waiter and313        resource model is defined.314        """315        waiter = WaiterAction(resource_waiter_model,316                              waiter_resource_name=resource_waiter_model.name)317        def do_waiter(self, *args, **kwargs):318            waiter(self, *args, **kwargs)319        do_waiter.__name__ = str(resource_waiter_model.name)320        do_waiter.__doc__ = docstring.ResourceWaiterDocstring(321            resource_name=resource_name,322            event_emitter=factory_self._emitter,323            service_model=service_context.service_model,324            resource_waiter_model=resource_waiter_model,325            service_waiter_model=service_context.service_waiter_model,326            include_signature=False327        )328        return do_waiter329    def _create_collection(factory_self, resource_name, collection_model,330                           service_context):331        """332        Creates a new property on the resource to lazy-load a collection.333        """334        cls = factory_self._collection_factory.load_from_definition(335            resource_name=resource_name, collection_model=collection_model,336            service_context=service_context,337            event_emitter=factory_self._emitter)338        def get_collection(self):339            return cls(340                collection_model=collection_model, parent=self,341                factory=factory_self, service_context=service_context)342        get_collection.__name__ = str(collection_model.name)343        get_collection.__doc__ = docstring.CollectionDocstring(344            collection_model=collection_model, include_signature=False)345        return property(get_collection)346    def _create_reference(factory_self, reference_model, resource_name,347                          service_context):348        """349        Creates a new property on the resource to lazy-load a reference.350        """351        # References are essentially an action with no request352        # or response, so we can re-use the response handlers to353        # build up resources from identifiers and data members.354        handler = ResourceHandler(355            search_path=reference_model.resource.path, factory=factory_self,356            resource_model=reference_model.resource,357            service_context=service_context358        )359        # Are there any identifiers that need access to data members?360        # This is important when building the resource below since361        # it requires the data to be loaded.362        needs_data = any(i.source == 'data' for i in363                         reference_model.resource.identifiers)364        def get_reference(self):365            # We need to lazy-evaluate the reference to handle circular366            # references between resources. We do this by loading the class367            # when first accessed.368            # This is using a *response handler* so we need to make sure369            # our data is loaded (if possible) and pass that data into370            # the handler as if it were a response. This allows references371            # to have their data loaded properly.372            if needs_data and self.meta.data is None and hasattr(self, 'load'):373                self.load()374            return handler(self, {}, self.meta.data)375        get_reference.__name__ = str(reference_model.name)376        get_reference.__doc__ = docstring.ReferenceDocstring(377            reference_model=reference_model,378            include_signature=False379        )380        return property(get_reference)381    def _create_class_partial(factory_self, subresource_model, resource_name,382                              service_context):383        """384        Creates a new method which acts as a functools.partial, passing385        along the instance's low-level `client` to the new resource386        class' constructor.387        """388        name = subresource_model.resource.type389        def create_resource(self, *args, **kwargs):390            # We need a new method here because we want access to the391            # instance's client.392            positional_args = []393            # We lazy-load the class to handle circular references.394            json_def = service_context.resource_json_definitions.get(name, {})395            resource_cls = factory_self.load_from_definition(396                resource_name=name,397                single_resource_json_definition=json_def,398                service_context=service_context399            )400            # Assumes that identifiers are in order, which lets you do401            # e.g. ``sqs.Queue('foo').Message('bar')`` to create a new message402            # linked with the ``foo`` queue and which has a ``bar`` receipt403            # handle. If we did kwargs here then future positional arguments404            # would lead to failure.405            identifiers = subresource_model.resource.identifiers406            if identifiers is not None:407                for identifier, value in build_identifiers(identifiers, self):408                    positional_args.append(value)409            return partial(resource_cls, *positional_args,410                           client=self.meta.client)(*args, **kwargs)411        create_resource.__name__ = str(name)412        create_resource.__doc__ = docstring.SubResourceDocstring(413            resource_name=resource_name,414            sub_resource_model=subresource_model,415            service_model=service_context.service_model,416            include_signature=False417        )418        return create_resource419    def _create_action(factory_self, action_model, resource_name,420                       service_context, is_load=False):421        """422        Creates a new method which makes a request to the underlying423        AWS service.424        """425        # Create the action in in this closure but before the ``do_action``426        # method below is invoked, which allows instances of the resource427        # to share the ServiceAction instance.428        action = ServiceAction(429            action_model, factory=factory_self,430            service_context=service_context431        )432        # A resource's ``load`` method is special because it sets433        # values on the resource instead of returning the response.434        if is_load:435            # We need a new method here because we want access to the436            # instance via ``self``.437            def do_action(self, *args, **kwargs):438                response = action(self, *args, **kwargs)439                self.meta.data = response440            # Create the docstring for the load/reload mehtods.441            lazy_docstring = docstring.LoadReloadDocstring(442                action_name=action_model.name,443                resource_name=resource_name,444                event_emitter=factory_self._emitter,445                load_model=action_model,446                service_model=service_context.service_model,447                include_signature=False448            )449        else:450            # We need a new method here because we want access to the451            # instance via ``self``.452            def do_action(self, *args, **kwargs):453                response = action(self, *args, **kwargs)454                if hasattr(self, 'load'):455                    # Clear cached data. It will be reloaded the next456                    # time that an attribute is accessed.457                    # TODO: Make this configurable in the future?458                    self.meta.data = None459                return response460            lazy_docstring = docstring.ActionDocstring(461                resource_name=resource_name,462                event_emitter=factory_self._emitter,463                action_model=action_model,464                service_model=service_context.service_model,465                include_signature=False466            )467        do_action.__name__ = str(action_model.name)468        do_action.__doc__ = lazy_docstring...

resource_registry.py

Source:resource_registry.py

1#    Licensed under the Apache License, Version 2.0 (the "License"); you may2#    not use this file except in compliance with the License. You may obtain3#    a copy of the License at4#5#         http://www.apache.org/licenses/LICENSE-2.06#7#    Unless required by applicable law or agreed to in writing, software8#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT9#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the10#    License for the specific language governing permissions and limitations11#    under the License.12from oslo_config import cfg13from oslo_log import log14import six15from neutron._i18n import _, _LI, _LW16from neutron.quota import resource17LOG = log.getLogger(__name__)18# Wrappers for easing access to the ResourceRegistry singleton19def register_resource(resource):20    ResourceRegistry.get_instance().register_resource(resource)21def register_resource_by_name(resource_name, plural_name=None):  # ä¾å¦resource_nameä¸ºnetwork22    ResourceRegistry.get_instance().register_resource_by_name(23        resource_name, plural_name)24def get_all_resources():25    return ResourceRegistry.get_instance().resources26def unregister_all_resources():27    if not ResourceRegistry._instance:28        return29    return ResourceRegistry.get_instance().unregister_resources()30def get_resource(resource_name):31    return ResourceRegistry.get_instance().get_resource(resource_name)32def is_tracked(resource_name):33    return ResourceRegistry.get_instance().is_tracked(resource_name)34# auxiliary functions and decorators35def set_resources_dirty(context):36    """Sets the dirty bit for resources with usage changes.37    This routine scans all registered resources, and, for those whose38    dirty status is True, sets the dirty bit to True in the database39    for the appropriate tenants.40    Please note that this routine begins a nested transaction, and it41    is not recommended that this transaction begins within another42    transaction. For this reason the function will raise a SqlAlchemy43    exception if such an attempt is made.44    :param context: a Neutron request context with a DB session45    """46    if not cfg.CONF.QUOTAS.track_quota_usage:47        return48    for res in get_all_resources().values():49        with context.session.begin(subtransactions=True):50            if is_tracked(res.name) and res.dirty:51                res.mark_dirty(context)52def resync_resource(context, resource_name, tenant_id):53    if not cfg.CONF.QUOTAS.track_quota_usage:54        return55    if is_tracked(resource_name):56        res = get_resource(resource_name)57        # If the resource is tracked count supports the resync_usage parameter58        res.resync(context, tenant_id)59def mark_resources_dirty(f):60    """Decorator for functions which alter resource usage.61    This decorator ensures set_resource_dirty is invoked after completion62    of the decorated function.63    """64    @six.wraps(f)65    def wrapper(_self, context, *args, **kwargs):66        ret_val = f(_self, context, *args, **kwargs)67        set_resources_dirty(context)68        return ret_val69    return wrapper70class tracked_resources(object):71    """Decorator for specifying resources for which usage should be tracked.72    A plugin class can use this decorator to specify for which resources73    usage info should be tracked into an appropriate table rather than being74    explicitly counted.75    """76    def __init__(self, override=False, **kwargs):77        self._tracked_resources = kwargs78        self._override = override79    def __call__(self, f):80        @six.wraps(f)81        def wrapper(*args, **kwargs):82            registry = ResourceRegistry.get_instance()83            for resource_name in self._tracked_resources:84                registry.set_tracked_resource(85                    resource_name,86                    self._tracked_resources[resource_name],87                    self._override)88            return f(*args, **kwargs)89        return wrapper90class ResourceRegistry(object):91    """Registry for resource subject to quota limits.92    This class keeps track of Neutron resources for which quota limits are93    enforced, regardless of whether their usage is being tracked or counted.94    For tracked-usage resources, that is to say those resources for which95    there are usage counters which are kept in sync with the actual number96    of rows in the database, this class allows the plugin to register their97    names either explicitly or through the @tracked_resources decorator,98    which should preferably be applied to the __init__ method of the class.99    """100    _instance = None101    @classmethod102    def get_instance(cls):103        if cls._instance is None:104            cls._instance = cls()105        return cls._instance106    def __init__(self):107        self._resources = {}108        # Map usage tracked resources to the correspondent db model class109        self._tracked_resource_mappings = {}110    def __contains__(self, resource):111        return resource in self._resources112    def _create_resource_instance(self, resource_name, plural_name):113        """Factory function for quota Resource.114        This routine returns a resource instance of the appropriate type115        according to system configuration.116        If QUOTAS.track_quota_usage is True, and there is a model mapping for117        the current resource, this function will return an instance of118        AccountedResource; otherwise an instance of CountableResource.119        # éé¢èµæºå·¥å120        # æ¤ä¾ç¨æ ¹æ®ç³»ç»éç½®è¿åéå½ç±»åçèµæºå®ä¾121        # å¦æQUOTAS.track_quota_usageä¸ºç,å¹¶ä¸åå¨å½åèµæºçæ¨¡åæ å°ï¼è¿ä¸ªå½æ°å°è¿åä¸ä¸ªAccountedResourceçå®ä¾122        # å¦åè¿åä¸ä¸ªCountableResourceçå®ä¾123        """124        if (not cfg.CONF.QUOTAS.track_quota_usage or125            resource_name not in self._tracked_resource_mappings):126            LOG.info(_LI("Creating instance of CountableResource for "127                         "resource:%s"), resource_name)128            return resource.CountableResource(129                resource_name, resource._count_resource,130                'quota_%s' % resource_name)131        else:132            LOG.info(_LI("Creating instance of TrackedResource for "133                         "resource:%s"), resource_name)134            return resource.TrackedResource(135                resource_name,136                self._tracked_resource_mappings[resource_name],137                'quota_%s' % resource_name)138    def set_tracked_resource(self, resource_name, model_class, override=False):139        # Do not do anything if tracking is disabled by config140        if not cfg.CONF.QUOTAS.track_quota_usage:141            return142        if isinstance(self._resources.get(resource_name),143                      resource.CountableResource):144            raise RuntimeError(_("Resource %s is already registered as a "145                                 "countable resource.") % resource_name)146        current_model_class = self._tracked_resource_mappings.setdefault(147            resource_name, model_class)148        # Check whether setdefault also set the entry in the dict149        if current_model_class != model_class:150            LOG.debug("A model class is already defined for %(resource)s: "151                      "%(current_model_class)s. Override:%(override)s",152                      {'resource': resource_name,153                       'current_model_class': current_model_class,154                       'override': override})155            if override:156                self._tracked_resource_mappings[resource_name] = model_class157        LOG.debug("Tracking information for resource: %s configured",158                  resource_name)159    def is_tracked(self, resource_name):160        """Find out if a resource if tracked or not.161        :param resource_name: name of the resource.162        :returns True if resource_name is registered and tracked, otherwise163                 False. Please note that here when False it returned it164                 simply means that resource_name is not a TrackedResource165                 instance, it does not necessarily mean that the resource166                 is not registered.167        """168        return resource_name in self._tracked_resource_mappings169    def register_resource(self, resource):170        if resource.name in self._resources:171            LOG.warning(_LW('%s is already registered'), resource.name)172        if resource.name in self._tracked_resource_mappings:173            resource.register_events()174        self._resources[resource.name] = resource175    def register_resources(self, resources):176        for res in resources:177            self.register_resource(res)178    def register_resource_by_name(self, resource_name,  # ä¾å¦resource_nameä¸ºnetwork179                                  plural_name=None):180        """Register a resource by name."""181        resource = self._create_resource_instance(182            resource_name, plural_name)  # CountableResource æèæ¯TrackedResource183        self.register_resource(resource)184    def unregister_resources(self):185        """Unregister all resources."""186        for (res_name, res) in self._resources.items():187            if res_name in self._tracked_resource_mappings:188                res.unregister_events()189        self._resources.clear()190        self._tracked_resource_mappings.clear()191    def get_resource(self, resource_name):192        """Return a resource given its name.193        :returns: The resource instance or None if the resource is not found194        """195        return self._resources.get(resource_name)196    @property197    def resources(self):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.