Coverage for typed_stream/streamable.py: 90%

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"""Streamable Interface providing a stream method.""" 

6 

7from __future__ import annotations 

8 

9from abc import ABC 

10from collections.abc import Iterable 

11from typing import TYPE_CHECKING, Literal, SupportsIndex, TypeVar, overload 

12 

13from ._impl._typing import override 

14 

15if TYPE_CHECKING: # pragma: no cover 

16 from ._impl.stream import Stream 

17 

18__all__: tuple[Literal["Streamable"], Literal["StreamableSequence"]] = ( 

19 "Streamable", 

20 "StreamableSequence", 

21) 

22 

23T = TypeVar("T") 

24V = TypeVar("V") 

25 

26 

27class Streamable(Iterable[T], ABC): 

28 """Abstract base class defining a Streamable interface.""" 

29 

30 __slots__ = () 

31 

32 def stream(self) -> Stream[T]: 

33 """Return Stream(self).""" 

34 from ._impl.stream import ( # pylint: disable=import-outside-toplevel 

35 Stream, 

36 ) 

37 

38 return Stream(self) 

39 

40 

41class StreamableSequence(tuple[T, ...], Streamable[T]): 

42 """A streamable immutable Sequence.""" 

43 

44 __slots__ = () 

45 

46 @overload 

47 def __add__(self, other: tuple[T, ...], /) -> StreamableSequence[T]: 

48 """Nobody inspects the spammish repetition.""" 

49 

50 @overload 

51 def __add__( 

52 self, other: tuple[V, ...], / # noqa: W504 

53 ) -> StreamableSequence[T | V]: 

54 """Nobody inspects the spammish repetition.""" 

55 

56 @override 

57 def __add__( 

58 self, other: tuple[T | V, ...], / # noqa: W504 

59 ) -> StreamableSequence[T | V]: 

60 """Add another StreamableSequence to this.""" 

61 if (result := super().__add__(other)) is NotImplemented: 

62 return NotImplemented 

63 if isinstance(result, StreamableSequence): 

64 return result 

65 return StreamableSequence(result) 

66 

67 @override 

68 def __mul__(self, other: SupportsIndex, /) -> StreamableSequence[T]: 

69 """Repeat self.""" 

70 return StreamableSequence(super().__mul__(other)) 

71 

72 @override 

73 def __rmul__(self, other: SupportsIndex, /) -> StreamableSequence[T]: 

74 """Repeat self.""" 

75 return StreamableSequence(super().__rmul__(other)) 

76 

77 @overload 

78 def __getitem__(self, item: SupportsIndex, /) -> T: 

79 """Nobody inspects the spammish repetition.""" 

80 

81 @overload 

82 def __getitem__( 

83 self, item: slice[int | None, int | None, int | None], / # noqa: W504 

84 ) -> StreamableSequence[T]: 

85 """Nobody inspects the spammish repetition.""" 

86 

87 @override 

88 def __getitem__( 

89 self, 

90 item: slice[int | None, int | None, int | None] | SupportsIndex, 

91 /, # noqa: W504 

92 ) -> StreamableSequence[T] | T: 

93 """Finish the stream by collecting.""" 

94 if isinstance(item, slice): 

95 return StreamableSequence(super().__getitem__(item)) 

96 return super().__getitem__(item)