Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/pubsub_v1/services/subscriber/pagers.py: 34%

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

2# Copyright 2022 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 Any, 

18 AsyncIterator, 

19 Awaitable, 

20 Callable, 

21 Sequence, 

22 Tuple, 

23 Optional, 

24 Iterator, 

25) 

26 

27from google.pubsub_v1.types import pubsub 

28 

29 

30class ListSubscriptionsPager: 

31 """A pager for iterating through ``list_subscriptions`` requests. 

32 

33 This class thinly wraps an initial 

34 :class:`google.pubsub_v1.types.ListSubscriptionsResponse` object, and 

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

36 ``subscriptions`` field. 

37 

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

39 ``ListSubscriptions`` requests and continue to iterate 

40 through the ``subscriptions`` field on the 

41 corresponding responses. 

42 

43 All the usual :class:`google.pubsub_v1.types.ListSubscriptionsResponse` 

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

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

46 """ 

47 

48 def __init__( 

49 self, 

50 method: Callable[..., pubsub.ListSubscriptionsResponse], 

51 request: pubsub.ListSubscriptionsRequest, 

52 response: pubsub.ListSubscriptionsResponse, 

53 *, 

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

55 ): 

56 """Instantiate the pager. 

57 

58 Args: 

59 method (Callable): The method that was originally called, and 

60 which instantiated this pager. 

61 request (google.pubsub_v1.types.ListSubscriptionsRequest): 

62 The initial request object. 

63 response (google.pubsub_v1.types.ListSubscriptionsResponse): 

64 The initial response object. 

65 metadata (Sequence[Tuple[str, str]]): Strings which should be 

66 sent along with the request as metadata. 

67 """ 

68 self._method = method 

69 self._request = pubsub.ListSubscriptionsRequest(request) 

70 self._response = response 

71 self._metadata = metadata 

72 

73 def __getattr__(self, name: str) -> Any: 

74 return getattr(self._response, name) 

75 

76 @property 

77 def pages(self) -> Iterator[pubsub.ListSubscriptionsResponse]: 

78 yield self._response 

79 while self._response.next_page_token: 

80 self._request.page_token = self._response.next_page_token 

81 self._response = self._method(self._request, metadata=self._metadata) 

82 yield self._response 

83 

84 def __iter__(self) -> Iterator[pubsub.Subscription]: 

85 for page in self.pages: 

86 yield from page.subscriptions 

87 

88 def __repr__(self) -> str: 

89 return "{0}<{1!r}>".format(self.__class__.__name__, self._response) 

90 

91 

92class ListSubscriptionsAsyncPager: 

93 """A pager for iterating through ``list_subscriptions`` requests. 

94 

95 This class thinly wraps an initial 

96 :class:`google.pubsub_v1.types.ListSubscriptionsResponse` object, and 

97 provides an ``__aiter__`` method to iterate through its 

98 ``subscriptions`` field. 

99 

100 If there are more pages, the ``__aiter__`` method will make additional 

101 ``ListSubscriptions`` requests and continue to iterate 

102 through the ``subscriptions`` field on the 

103 corresponding responses. 

104 

105 All the usual :class:`google.pubsub_v1.types.ListSubscriptionsResponse` 

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

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

108 """ 

109 

110 def __init__( 

111 self, 

112 method: Callable[..., Awaitable[pubsub.ListSubscriptionsResponse]], 

113 request: pubsub.ListSubscriptionsRequest, 

114 response: pubsub.ListSubscriptionsResponse, 

115 *, 

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

117 ): 

118 """Instantiates the pager. 

119 

120 Args: 

121 method (Callable): The method that was originally called, and 

122 which instantiated this pager. 

123 request (google.pubsub_v1.types.ListSubscriptionsRequest): 

124 The initial request object. 

125 response (google.pubsub_v1.types.ListSubscriptionsResponse): 

126 The initial response object. 

127 metadata (Sequence[Tuple[str, str]]): Strings which should be 

128 sent along with the request as metadata. 

129 """ 

130 self._method = method 

131 self._request = pubsub.ListSubscriptionsRequest(request) 

132 self._response = response 

133 self._metadata = metadata 

134 

135 def __getattr__(self, name: str) -> Any: 

136 return getattr(self._response, name) 

137 

138 @property 

139 async def pages(self) -> AsyncIterator[pubsub.ListSubscriptionsResponse]: 

140 yield self._response 

141 while self._response.next_page_token: 

142 self._request.page_token = self._response.next_page_token 

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

144 yield self._response 

145 

146 def __aiter__(self) -> AsyncIterator[pubsub.Subscription]: 

147 async def async_generator(): 

148 async for page in self.pages: 

149 for response in page.subscriptions: 

150 yield response 

151 

152 return async_generator() 

153 

154 def __repr__(self) -> str: 

155 return "{0}<{1!r}>".format(self.__class__.__name__, self._response) 

156 

157 

158class ListSnapshotsPager: 

159 """A pager for iterating through ``list_snapshots`` requests. 

160 

161 This class thinly wraps an initial 

162 :class:`google.pubsub_v1.types.ListSnapshotsResponse` object, and 

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

164 ``snapshots`` field. 

165 

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

167 ``ListSnapshots`` requests and continue to iterate 

168 through the ``snapshots`` field on the 

169 corresponding responses. 

170 

171 All the usual :class:`google.pubsub_v1.types.ListSnapshotsResponse` 

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

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

174 """ 

175 

176 def __init__( 

177 self, 

178 method: Callable[..., pubsub.ListSnapshotsResponse], 

179 request: pubsub.ListSnapshotsRequest, 

180 response: pubsub.ListSnapshotsResponse, 

181 *, 

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

183 ): 

184 """Instantiate the pager. 

185 

186 Args: 

187 method (Callable): The method that was originally called, and 

188 which instantiated this pager. 

189 request (google.pubsub_v1.types.ListSnapshotsRequest): 

190 The initial request object. 

191 response (google.pubsub_v1.types.ListSnapshotsResponse): 

192 The initial response object. 

193 metadata (Sequence[Tuple[str, str]]): Strings which should be 

194 sent along with the request as metadata. 

195 """ 

196 self._method = method 

197 self._request = pubsub.ListSnapshotsRequest(request) 

198 self._response = response 

199 self._metadata = metadata 

200 

201 def __getattr__(self, name: str) -> Any: 

202 return getattr(self._response, name) 

203 

204 @property 

205 def pages(self) -> Iterator[pubsub.ListSnapshotsResponse]: 

206 yield self._response 

207 while self._response.next_page_token: 

208 self._request.page_token = self._response.next_page_token 

209 self._response = self._method(self._request, metadata=self._metadata) 

210 yield self._response 

211 

212 def __iter__(self) -> Iterator[pubsub.Snapshot]: 

213 for page in self.pages: 

214 yield from page.snapshots 

215 

216 def __repr__(self) -> str: 

217 return "{0}<{1!r}>".format(self.__class__.__name__, self._response) 

218 

219 

220class ListSnapshotsAsyncPager: 

221 """A pager for iterating through ``list_snapshots`` requests. 

222 

223 This class thinly wraps an initial 

224 :class:`google.pubsub_v1.types.ListSnapshotsResponse` object, and 

225 provides an ``__aiter__`` method to iterate through its 

226 ``snapshots`` field. 

227 

228 If there are more pages, the ``__aiter__`` method will make additional 

229 ``ListSnapshots`` requests and continue to iterate 

230 through the ``snapshots`` field on the 

231 corresponding responses. 

232 

233 All the usual :class:`google.pubsub_v1.types.ListSnapshotsResponse` 

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

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

236 """ 

237 

238 def __init__( 

239 self, 

240 method: Callable[..., Awaitable[pubsub.ListSnapshotsResponse]], 

241 request: pubsub.ListSnapshotsRequest, 

242 response: pubsub.ListSnapshotsResponse, 

243 *, 

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

245 ): 

246 """Instantiates the pager. 

247 

248 Args: 

249 method (Callable): The method that was originally called, and 

250 which instantiated this pager. 

251 request (google.pubsub_v1.types.ListSnapshotsRequest): 

252 The initial request object. 

253 response (google.pubsub_v1.types.ListSnapshotsResponse): 

254 The initial response object. 

255 metadata (Sequence[Tuple[str, str]]): Strings which should be 

256 sent along with the request as metadata. 

257 """ 

258 self._method = method 

259 self._request = pubsub.ListSnapshotsRequest(request) 

260 self._response = response 

261 self._metadata = metadata 

262 

263 def __getattr__(self, name: str) -> Any: 

264 return getattr(self._response, name) 

265 

266 @property 

267 async def pages(self) -> AsyncIterator[pubsub.ListSnapshotsResponse]: 

268 yield self._response 

269 while self._response.next_page_token: 

270 self._request.page_token = self._response.next_page_token 

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

272 yield self._response 

273 

274 def __aiter__(self) -> AsyncIterator[pubsub.Snapshot]: 

275 async def async_generator(): 

276 async for page in self.pages: 

277 for response in page.snapshots: 

278 yield response 

279 

280 return async_generator() 

281 

282 def __repr__(self) -> str: 

283 return "{0}<{1!r}>".format(self.__class__.__name__, self._response)