Coverage for faststream / redis / schemas / stream_sub.py: 91%

1import warnings (empty)

2from copy import deepcopy (empty)

3 

4from faststream._internal.proto import NameRequired (empty)

5from faststream.exceptions import SetupError (empty)

6 

7 

8class StreamSub(NameRequired): (empty)

9 """A class to represent a Redis Stream subscriber. (empty)

10 

11 Args: 

12 batch: 

13 Whether to consume messages in batches or not. 

14 max_records: 

15 Number of messages to consume as one batch. 

16 consumer: 

17 The consumer unique name 

18 

19 https://redis.io/docs/latest/develop/tools/insight/tutorials/insight-stream-consumer/#run-the-consumer 

20 group: 

21 The name of consumer group 

22 last_id: 

23 An Entry ID, which uses to pick up from where it left off after it is restarted. 

24 maxlen: 

25 Redis Stream maxlen publish option. Remove eldest message if maxlen exceeded. 

26 

27 https://redis.io/docs/latest/develop/data-types/streams/#capped-streams 

28 name: 

29 The original Redis Stream name. 

30 no_ack: 

31 If True, to enable the XREADGROUP NOACK subcommand. 

32 

33 https://redis.io/docs/latest/commands/xreadgroup/#differences-between-xread-and-xreadgroup 

34 polling_interval: 

35 Polling interval in milliseconds. 

36 min_idle_time: 

37 Minimum idle time in milliseconds for a message to be eligible for claiming via XAUTOCLAIM. 

38 Messages that have been pending (unacknowledged) for at least this duration can be 

39 reclaimed by this consumer. Only applicable when using consumer groups. 

40 

41 https://redis.io/docs/latest/commands/xautoclaim/ 

42 """ 

43 

44 __slots__ = ( (empty)

45 "batch", 

46 "consumer", 

47 "group", 

48 "last_id", 

49 "max_records", 

50 "maxlen", 

51 "min_idle_time", 

52 "name", 

53 "no_ack", 

54 "polling_interval", 

55 ) 

56 

57 def __init__( (empty)

58 self, 

59 stream: str, 

60 polling_interval: int | None = None, 

61 group: str | None = None, 

62 consumer: str | None = None, 

63 batch: bool = False, 

64 no_ack: bool = False, 

65 last_id: str | None = None, 

66 maxlen: int | None = None, 

67 max_records: int | None = None, 

68 min_idle_time: int | None = None, 

69 ) -> None: 

70 if (group and not consumer) or (not group and consumer): (empty)

71 msg = "You should specify `group` and `consumer` both" (empty)

72 raise SetupError(msg) (empty)

73 

74 if last_id is None: (empty)

75 last_id = ">" if group and consumer else "$" (empty)

76 

77 if group and consumer: (empty)

78 if last_id != ">": (empty)

79 if polling_interval: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true(empty)

80 warnings.warn( 

81 message="`polling_interval` is not supported by consumer group with last_id other than `>`", 

82 category=RuntimeWarning, 

83 stacklevel=1, 

84 ) 

85 

86 if no_ack: (empty)

87 warnings.warn( (empty)

88 message="`no_ack` is not supported by consumer group with last_id other than `>`", 

89 category=RuntimeWarning, 

90 stacklevel=1, 

91 ) 

92 

93 elif no_ack: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true(empty)

94 warnings.warn( 

95 message="`no_ack` has no effect with consumer group", 

96 category=RuntimeWarning, 

97 stacklevel=1, 

98 ) 

99 

100 super().__init__(stream) (empty)

101 

102 self.group = group (empty)

103 self.consumer = consumer (empty)

104 self.polling_interval = polling_interval or 100 (empty)

105 self.batch = batch (empty)

106 self.no_ack = no_ack (empty)

107 self.last_id = last_id (empty)

108 self.maxlen = maxlen (empty)

109 self.max_records = max_records (empty)

110 self.min_idle_time = min_idle_time (empty)

111 

112 def add_prefix(self, prefix: str) -> "StreamSub": (empty)

113 new_stream = deepcopy(self) (empty)

114 new_stream.name = f"{prefix}{new_stream.name}" (empty)

115 return new_stream (empty)