Coverage for src/gitlabracadabra/packages/stream.py: 90%

1# 

2# Copyright (C) 2019-2025 Mathieu Parent <math.parent@gmail.com> 

3# 

4# This program is free software: you can redistribute it and/or modify 

5# it under the terms of the GNU Lesser General Public License as published by 

6# the Free Software Foundation, either version 3 of the License, or 

7# (at your option) any later version. 

8# 

9# This program is distributed in the hope that it will be useful, 

10# but WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

12# GNU Lesser General Public License for more details. 

13# 

14# You should have received a copy of the GNU Lesser General Public License 

15# along with this program. If not, see <http://www.gnu.org/licenses/>. 

16 

17from __future__ import annotations 

18 

19from typing import TYPE_CHECKING, AnyStr 

20 

21if TYPE_CHECKING: 

22 from collections.abc import Iterator 

23 

24 from requests.models import Response 

25 

26 

27class Stream: 

28 """Stream.""" 

29 

30 def __init__(self, response: Response, chunksize: int = 65536) -> None: 

31 """Initialize Stream. 

32 

33 Args: 

34 response: Streamed response. 

35 chunksize: Chunk size (used when there is no Content-Length header). 

36 """ 

37 self._response = response 

38 self._chunksize = chunksize 

39 

40 def __bool__(self) -> bool: 

41 """Stream as boolean. 

42 

43 Needed for Session.request() which uses: data=data or dict(). 

44 (otherwise, would be considered False when length is 0). 

45 

46 Returns: 

47 Always True. 

48 """ 

49 return True 

50 

51 def __len__(self) -> int: 

52 """Get stream length. 

53 

54 Returns: 

55 The stream length. Zero if there is no Content-Length header. 

56 """ 

57 return int(self._response.headers.get("Content-Length", "0")) 

58 

59 def __iter__(self) -> Iterator[bytes]: 

60 """Get an iterator of chunks of body. 

61 

62 Returns: 

63 A bytes iterator. 

64 """ 

65 return self._response.raw.stream(self._chunksize) 

66 

67 @property 

68 def name(self) -> str: 

69 """Return URL. 

70 

71 This is needed to have proper file name in multipart upload. 

72 

73 Called from requests.utils.guess_filename(), 

74 called from requests.models.RequestEncodingMixin._encode_files(), 

75 called from requests.models.PreparedRequest.prepare_body(). 

76 

77 Returns: 

78 The response URL. 

79 """ 

80 if self._response.history: 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was never true

81 # Keep original request URL, to avoid too long filename 

82 return self._response.history[0].url 

83 return self._response.url 

84 

85 def read(self, size: int | None = None) -> AnyStr: # type: ignore 

86 """Read stream. 

87 

88 Args: 

89 size: Length to read. Defaulting to None like http.client.HTTPResponse.read(). 

90 

91 Returns: 

92 The read bytes/str. 

93 """ 

94 return self._response.raw.read(size) # type: ignore