Coverage for faststream / nats / schemas / obj_watch.py: 89%

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

2 

3 

4class ObjWatch: (empty)

5 """A class to represent a NATS object storage watch subscription. 

6 

7 Args: 

8 ignore_deletes (bool): Ignore delete events (default is `False`). 

9 include_history (bool): Include history (default is `False`). 

10 meta_only (bool): Only metadata. (default is `False`). 

11 timeout (float): The timeout for the watch in seconds (default is `5.0`). 

12 declare (bool): Whether to create object storage automatically or just connect to it (default is `True`). 

13 """ 

14 

15 __slots__ = ( (empty)

16 "declare", 

17 "ignore_deletes", 

18 "include_history", 

19 "meta_only", 

20 "timeout", 

21 ) 

22 

23 def __init__( (empty)

24 self, 

25 ignore_deletes: bool = False, 

26 include_history: bool = False, 

27 meta_only: bool = False, 

28 timeout: float = 5.0, 

29 # custom 

30 declare: bool = True, 

31 ) -> None: 

32 self.ignore_deletes = ignore_deletes (empty)

33 self.include_history = include_history (empty)

34 self.meta_only = meta_only (empty)

35 self.timeout = timeout (empty)

36 

37 self.declare = declare (empty)

38 

39 @overload (empty)

40 @classmethod (empty)

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

42 

43 @overload (empty)

44 @classmethod (empty)

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

46 

47 @overload (empty)

48 @classmethod (empty)

49 def validate(cls, value: "ObjWatch") -> "ObjWatch": ... (empty)

50 

51 @overload (empty)

52 @classmethod (empty)

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

54 

55 @classmethod (empty)

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

57 if value is True: (empty)

58 return ObjWatch() (empty)

59 if value is False: 59 ↛ 61line 59 didn't jump to line 61 because the condition on line 59 was always true(empty)

60 return None (empty)

61 return value