Coverage for typed_stream/_impl/file_streams.py: 94%

1# Licensed under the EUPL-1.2 or later. 

2# You may obtain a copy of the licence in all the official languages of the 

3# European Union at https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12 

4 

5"""Stream classes for streaming file data.""" 

6 

7from __future__ import annotations 

8 

9from types import EllipsisType 

10from typing import AnyStr, Literal 

11 

12from ._lazy_file_iterators import ( 

13 LazyFileIterator, 

14 LazyFileIteratorRemovingEndsBytes, 

15 LazyFileIteratorRemovingEndsStr, 

16) 

17from ._types import PathLikeType 

18from ._typing import override 

19from .stream import Stream 

20 

21__all__: tuple[Literal["BinaryFileStream"], Literal["FileStream"]] = ( 

22 "BinaryFileStream", 

23 "FileStream", 

24) 

25 

26 

27class FileStreamBase(Stream[AnyStr]): 

28 """ABC for file streams.""" 

29 

30 _file_iterator: None | LazyFileIterator[AnyStr] 

31 __slots__ = ("_file_iterator",) 

32 

33 def _close_source(self) -> None: 

34 """Close the source of the Stream. Used in FileStream.""" 

35 if not self._file_iterator: 

36 return 

37 self._file_iterator.close() 

38 self._file_iterator = None 

39 

40 @override 

41 def _get_args(self) -> tuple[object, ...]: 

42 """Return the args used to initializing self.""" 

43 if not self._file_iterator: 

44 return (...,) 

45 

46 return ( 

47 self._file_iterator.path, 

48 self._file_iterator.encoding, 

49 # pylint: disable=unidiomatic-typecheck 

50 type(self._file_iterator) is LazyFileIterator, 

51 ) 

52 

53 

54class FileStream(FileStreamBase[str]): 

55 """Lazily iterate over a file.""" 

56 

57 __slots__ = () 

58 

59 def __init__( 

60 self, 

61 data: PathLikeType | EllipsisType, 

62 encoding: str = "UTF-8", 

63 keep_line_ends: bool = False, 

64 ) -> None: 

65 """Create a new FileStream. 

66 

67 To create a finished FileStream do FileStream(...). 

68 """ 

69 if isinstance(data, EllipsisType): 

70 self._file_iterator = None 

71 super().__init__(...) 

72 return 

73 

74 self._file_iterator = ( 

75 LazyFileIterator(data, encoding=encoding) 

76 if keep_line_ends 

77 else LazyFileIteratorRemovingEndsStr(data, encoding=encoding) 

78 ) 

79 super().__init__(self._file_iterator, self._close_source) 

80 

81 @classmethod 

82 @override 

83 def _module(cls) -> str: 

84 if cls == FileStream: 

85 return "typed_stream" 

86 return cls.__module__ 

87 

88 

89class BinaryFileStream(FileStreamBase[bytes]): 

90 """Lazily iterate over the lines of a file.""" 

91 

92 __slots__ = () 

93 

94 def __init__( 

95 self, 

96 data: PathLikeType | EllipsisType, 

97 keep_line_ends: bool = False, 

98 ) -> None: 

99 """Create a new BinaryFileStream. 

100 

101 To create a finished BinaryFileStream do BinaryFileStream(...). 

102 """ 

103 if isinstance(data, EllipsisType): 

104 self._file_iterator = None 

105 super().__init__(...) 

106 return 

107 

108 self._file_iterator = ( 

109 LazyFileIterator(data) 

110 if keep_line_ends 

111 else LazyFileIteratorRemovingEndsBytes(data) 

112 ) 

113 super().__init__(self._file_iterator, self._close_source) 

114 

115 @classmethod 

116 @override 

117 def _module(cls) -> str: 

118 if cls == BinaryFileStream: 

119 return "typed_stream" 

120 return cls.__module__