Coverage for faststream / redis / schemas / list_sub.py: 93%

1from copy import deepcopy (empty)

2from functools import cached_property (empty)

3 

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

5 

6 

7class ListSub(NameRequired): (empty)

8 """A class to represent a Redis List subscriber.""" 

9 

10 __slots__ = ( (empty)

11 "batch", 

12 "max_records", 

13 "name", 

14 "polling_interval", 

15 ) 

16 

17 def __init__( (empty)

18 self, 

19 list_name: str, 

20 batch: bool = False, 

21 max_records: int = 10, 

22 polling_interval: float = 0.1, 

23 ) -> None: 

24 super().__init__(list_name) (empty)

25 

26 self.batch = batch (empty)

27 self.max_records = max_records (empty)

28 self.polling_interval = polling_interval (empty)

29 

30 @cached_property (empty)

31 def records(self) -> int | None: (empty)

32 return self.max_records if self.batch else None 

33 

34 def add_prefix(self, prefix: str) -> "ListSub": (empty)

35 new_list = deepcopy(self) (empty)

36 new_list.name = f"{prefix}{new_list.name}" (empty)

37 return new_list (empty)