Coverage for faststream / nats / helpers / bucket_declarer.py: 90%

1from typing import TYPE_CHECKING, Optional (empty)

2 

3from nats.js.api import KeyValueConfig (empty)

4 

5from .state import ConnectedState, ConnectionState, EmptyConnectionState (empty)

6 

7if TYPE_CHECKING: (empty)

8 from nats.js import JetStreamContext 

9 from nats.js.api import Placement, RePublish, StorageType 

10 from nats.js.kv import KeyValue 

11 

12 

13class KVBucketDeclarer: (empty)

14 buckets: dict[str, "KeyValue"] (empty)

15 

16 def __init__(self) -> None: (empty)

17 self.buckets = {} (empty)

18 

19 self.__state: ConnectionState[JetStreamContext] = EmptyConnectionState() (empty)

20 

21 def connect(self, connection: "JetStreamContext") -> None: (empty)

22 self.__state = ConnectedState(connection) (empty)

23 

24 def disconnect(self) -> None: (empty)

25 self.__state = EmptyConnectionState() (empty)

26 

27 async def create_key_value( (empty)

28 self, 

29 bucket: str, 

30 *, 

31 description: str | None = None, 

32 max_value_size: int | None = None, 

33 history: int = 1, 

34 ttl: float | None = None, # in seconds 

35 max_bytes: int | None = None, 

36 storage: Optional["StorageType"] = None, 

37 replicas: int = 1, 

38 placement: Optional["Placement"] = None, 

39 republish: Optional["RePublish"] = None, 

40 direct: bool | None = None, 

41 # custom 

42 declare: bool = True, 

43 ) -> "KeyValue": 

44 if (key_value := self.buckets.get(bucket)) is None: (empty)

45 if declare: 45 ↛ 62line 45 didn't jump to line 62 because the condition on line 45 was always true(empty)

46 key_value = await self.__state.connection.create_key_value( (empty)

47 config=KeyValueConfig( 

48 bucket=bucket, 

49 description=description, 

50 max_value_size=max_value_size, 

51 history=history, 

52 ttl=ttl, 

53 max_bytes=max_bytes, 

54 storage=storage, 

55 replicas=replicas, 

56 placement=placement, 

57 republish=republish, 

58 direct=direct, 

59 ), 

60 ) 

61 else: 

62 key_value = await self.__state.connection.key_value(bucket) 

63 

64 self.buckets[bucket] = key_value (empty)

65 

66 return key_value (empty)