Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/rest_streaming.py: 42%

1# Copyright 2021 Google LLC 

2# 

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

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

5# You may obtain a copy of the License at 

6# 

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

8# 

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

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

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

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

13# limitations under the License. 

14 

15"""Helpers for server-side streaming in REST.""" 

16 

17from typing import Union 

18 

19import google.protobuf.message 

20import proto 

21import requests 

22 

23from google.api_core._rest_streaming_base import BaseResponseIterator 

24 

25 

26class ResponseIterator(BaseResponseIterator): 

27 """Iterator over REST API responses. 

28 

29 Args: 

30 response (requests.Response): An API response object. 

31 response_message_cls (Union[proto.Message, google.protobuf.message.Message]): A response 

32 class expected to be returned from an API. 

33 

34 Raises: 

35 ValueError: 

36 - If `response_message_cls` is not a subclass of `proto.Message` or `google.protobuf.message.Message`. 

37 """ 

38 

39 def __init__( 

40 self, 

41 response: requests.Response, 

42 response_message_cls: Union[proto.Message, google.protobuf.message.Message], 

43 ): 

44 self._response = response 

45 # Inner iterator over HTTP response's content. 

46 self._response_itr = self._response.iter_content(decode_unicode=True) 

47 super(ResponseIterator, self).__init__( 

48 response_message_cls=response_message_cls 

49 ) 

50 

51 def cancel(self): 

52 """Cancel existing streaming operation.""" 

53 self._response.close() 

54 

55 def __next__(self): 

56 while not self._ready_objs: 

57 try: 

58 chunk = next(self._response_itr) 

59 self._process_chunk(chunk) 

60 except StopIteration as e: 

61 if self._level > 0: 

62 raise ValueError("Unfinished stream: %s" % self._obj) 

63 raise e 

64 return self._grab() 

65 

66 def __iter__(self): 

67 return self