Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/operations_v1/pagers_async.py: 40%

1# -*- coding: utf-8 -*- 

2# Copyright 2024 Google LLC 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16from typing import ( 

17 Callable, 

18 AsyncIterator, 

19 Sequence, 

20 Tuple, 

21) 

22 

23from google.longrunning import operations_pb2 

24from google.api_core.operations_v1.pagers_base import ListOperationsPagerBase 

25 

26 

27class ListOperationsAsyncPager(ListOperationsPagerBase): 

28 """A pager for iterating through ``list_operations`` requests. 

29 

30 This class thinly wraps an initial 

31 :class:`google.longrunning.operations_pb2.ListOperationsResponse` object, and 

32 provides an ``__iter__`` method to iterate through its 

33 ``operations`` field. 

34 

35 If there are more pages, the ``__iter__`` method will make additional 

36 ``ListOperations`` requests and continue to iterate 

37 through the ``operations`` field on the 

38 corresponding responses. 

39 

40 All the usual :class:`google.longrunning.operations_pb2.ListOperationsResponse` 

41 attributes are available on the pager. If multiple requests are made, only 

42 the most recent response is retained, and thus used for attribute lookup. 

43 """ 

44 

45 def __init__( 

46 self, 

47 method: Callable[..., operations_pb2.ListOperationsResponse], 

48 request: operations_pb2.ListOperationsRequest, 

49 response: operations_pb2.ListOperationsResponse, 

50 *, 

51 metadata: Sequence[Tuple[str, str]] = () 

52 ): 

53 super().__init__( 

54 method=method, request=request, response=response, metadata=metadata 

55 ) 

56 

57 @property 

58 async def pages(self) -> AsyncIterator[operations_pb2.ListOperationsResponse]: 

59 yield self._response 

60 while self._response.next_page_token: 

61 self._request.page_token = self._response.next_page_token 

62 self._response = await self._method(self._request, metadata=self._metadata) 

63 yield self._response 

64 

65 def __aiter__(self) -> AsyncIterator[operations_pb2.Operation]: 

66 async def async_generator(): 

67 async for page in self.pages: 

68 for operation in page.operations: 

69 yield operation 

70 

71 return async_generator()