Coverage for faststream / nats / schemas / pull_sub.py: 88%

1from typing import Literal, Optional, Union, overload (empty)

2 

3 

4class PullSub: (empty)

5 """A class to represent a NATS pull subscription. 

6 

7 Args: 

8 batch_size (int): Consuming messages batch size. (default is `1`). 

9 timeout (:obj:`float`, optional): Wait this time for required batch size will be accumulated in stream 

10 in seconds (default is `5.0`). 

11 batch (bool): Whether to propagate consuming batch as iterable object to your handler (default is `False`). 

12 """ 

13 

14 __slots__ = ( (empty)

15 "batch", 

16 "batch_size", 

17 "timeout", 

18 ) 

19 

20 def __init__( (empty)

21 self, 

22 batch_size: int = 1, 

23 timeout: float | None = 5.0, 

24 batch: bool = False, 

25 ) -> None: 

26 self.batch_size = batch_size (empty)

27 self.batch = batch (empty)

28 self.timeout = timeout (empty)

29 

30 @overload (empty)

31 @classmethod (empty)

32 def validate(cls, value: Literal[True]) -> "PullSub": ... (empty)

33 

34 @overload (empty)

35 @classmethod (empty)

36 def validate(cls, value: Literal[False]) -> None: ... (empty)

37 

38 @overload (empty)

39 @classmethod (empty)

40 def validate(cls, value: "PullSub") -> "PullSub": ... (empty)

41 

42 @overload (empty)

43 @classmethod (empty)

44 def validate(cls, value: Union[bool, "PullSub"]) -> Optional["PullSub"]: ... (empty)

45 

46 @classmethod (empty)

47 def validate(cls, value: Union[bool, "PullSub"]) -> Optional["PullSub"]: (empty)

48 if value is True: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true(empty)

49 return PullSub() 

50 if value is False: (empty)

51 return None (empty)

52 return value (empty)