codekingpro/portable-devtools

1# --------------------------------------------------------------------------2#3# Copyright (c) Microsoft Corporation. All rights reserved.4#5# The MIT License (MIT)6#7# Permission is hereby granted, free of charge, to any person obtaining a copy8# of this software and associated documentation files (the ""Software""), to9# deal in the Software without restriction, including without limitation the10# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or11# sell copies of the Software, and to permit persons to whom the Software is12# furnished to do so, subject to the following conditions:13#14# The above copyright notice and this permission notice shall be included in15# all copies or substantial portions of the Software.16#17# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING22# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS23# IN THE SOFTWARE.24#25# --------------------------------------------------------------------------26import sys27try:28    from collections.abc import Iterator29    xrange = range30except ImportError:31    from collections import Iterator32 33from typing import Dict, Any, List, Callable, Optional, TYPE_CHECKING  # pylint: disable=unused-import34 35from .serialization import Deserializer36from .pipeline import ClientRawResponse37 38if TYPE_CHECKING:39    from .universal_http import ClientResponse  # pylint: disable=unused-import40    from .serialization import Model  # pylint: disable=unused-import41 42if sys.version_info >= (3, 5, 2):43    # Not executed on old Python, no syntax error44    from .async_paging import AsyncPagedMixin  # type: ignore45else:46    class AsyncPagedMixin(object):  # type: ignore47        pass48 49class Paged(AsyncPagedMixin, Iterator):50    """A container for paged REST responses.51 52    :param ClientResponse response: server response object.53    :param callable command: Function to retrieve the next page of items.54    :param dict classes: A dictionary of class dependencies for55     deserialization.56    :param dict raw_headers: A dict of raw headers to add if "raw" is called57    """58    _validation = {}  # type: Dict[str, Dict[str, Any]]59    _attribute_map = {}  # type: Dict[str, Dict[str, Any]]60 61    def __init__(self, command, classes, raw_headers=None, **kwargs):62        # type: (Callable[[str], ClientResponse], Dict[str, Model], Dict[str, str], Any) -> None63        super(Paged, self).__init__(**kwargs)  # type: ignore64        # Sets next_link, current_page, and _current_page_iter_index.65        self.next_link = ""66        self._current_page_iter_index = 067        self.reset()68        self._derserializer = Deserializer(classes)69        self._get_next = command70        self._response = None  # type: Optional[ClientResponse]71        self._raw_headers = raw_headers72 73    def __iter__(self):74        """Return 'self'."""75        # Since iteration mutates this object, consider it an iterator in-and-of76        # itself.77        return self78 79    @classmethod80    def _get_subtype_map(cls):81        """Required for parity to Model object for deserialization."""82        return {}83 84    @property85    def raw(self):86        # type: () -> ClientRawResponse87        """Get current page as ClientRawResponse.88 89        :rtype: ClientRawResponse90        """91        raw = ClientRawResponse(self.current_page, self._response)92        if self._raw_headers:93            raw.add_headers(self._raw_headers)94        return raw95 96    def get(self, url):97        # type: (str) -> List[Model]98        """Get an arbitrary page.99 100        This resets the iterator and then fully consumes it to return the101        specific page **only**.102 103        :param str url: URL to arbitrary page results.104        """105        self.reset()106        self.next_link = url107        return self.advance_page()108 109    def reset(self):110        # type: () -> None111        """Reset iterator to first page."""112        self.next_link = ""113        self.current_page = []  # type: List[Model]114        self._current_page_iter_index = 0115 116    def advance_page(self):117        # type: () -> List[Model]118        """Force moving the cursor to the next azure call.119 120        This method is for advanced usage, iterator protocol is preferred.121 122        :raises: StopIteration if no further page123        :return: The current page list124        :rtype: list125        """126        if self.next_link is None:127            raise StopIteration("End of paging")128        self._current_page_iter_index = 0129        self._response = self._get_next(self.next_link)130        self._derserializer(self, self._response)131        return self.current_page132 133    def __next__(self):134        """Iterate through responses."""135        # Storing the list iterator might work out better, but there's no136        # guarantee that some code won't replace the list entirely with a copy,137        # invalidating an list iterator that might be saved between iterations.138        if self.current_page and self._current_page_iter_index < len(self.current_page):139            response = self.current_page[self._current_page_iter_index]140            self._current_page_iter_index += 1141            return response142        else:143            self.advance_page()144            return self.__next__()145 146    next = __next__  # Python 2 compatibility.147