Coverage for archdocs/features/messaging_queue/const.py: 100%

1import dataclasses 

2import enum 

3import types 

4import typing 

5 

6 

7@typing.final 

8class BrokersEnum(enum.Enum): 

9 rabbit_broker = "rabbit" 

10 kafka_broker = "kafka" 

11 nats_broker = "nats" 

12 redis_broker = "redis" 

13 

14 

15BROKER_CLASS_OF_NAME: typing.Final = types.MappingProxyType( 

16 { 

17 BrokersEnum.rabbit_broker: "RabbitBroker", 

18 BrokersEnum.kafka_broker: "KafkaBroker", 

19 BrokersEnum.nats_broker: "NatsBroker", 

20 BrokersEnum.redis_broker: "RedisBroker", 

21 }, 

22) 

23DESTINATION_KEYWORDS: typing.Final = ("queue", "topic", "subject", "channel", "stream", "list") 

24 

25 

26@typing.final 

27class MessageDirection(enum.Enum): 

28 consumed = "consumed" 

29 produced = "produced" 

30 

31 

32@typing.final 

33@dataclasses.dataclass(slots=True, kw_only=True, frozen=True) 

34class BrokerFlow: 

35 broker_name: str 

36 consumes: bool 

37 produces: bool 

38 consumed_topics: tuple[str, ...] = () 

39 produced_topics: tuple[str, ...] = () 

40 

41 

42@typing.final 

43@dataclasses.dataclass(slots=True, kw_only=True, frozen=True) 

44class MQFeatures: 

45 broker_flows: tuple[BrokerFlow, ...] = ()