Coverage for faststream / kafka / publisher / state.py: 77%

1from abc import abstractmethod (empty)

2from typing import TYPE_CHECKING, Protocol (empty)

3 

4from faststream.exceptions import IncorrectState (empty)

5 

6if TYPE_CHECKING: (empty)

7 from aiokafka import AIOKafkaProducer 

8 

9 

10class ProducerState(Protocol): (empty)

11 producer: "AIOKafkaProducer" (empty)

12 

13 @property (empty)

14 @abstractmethod (empty)

15 def closed(self) -> bool: ... (empty)

16 

17 def __bool__(self) -> bool: ... (empty)

18 

19 async def stop(self) -> None: ... (empty)

20 

21 async def flush(self) -> None: ... (empty)

22 

23 

24class EmptyProducerState(ProducerState): (empty)

25 __slots__ = () (empty)

26 

27 closed = True (empty)

28 

29 @property (empty)

30 def producer(self) -> "AIOKafkaProducer": (empty)

31 msg = "You can't use producer here, please connect broker first." 

32 raise IncorrectState(msg) 

33 

34 def __bool__(self) -> bool: (empty)

35 return False 

36 

37 async def stop(self) -> None: (empty)

38 pass (empty)

39 

40 async def flush(self) -> None: (empty)

41 pass 

42 

43 

44class RealProducer(ProducerState): (empty)

45 __slots__ = ("producer",) (empty)

46 

47 def __init__(self, producer: "AIOKafkaProducer") -> None: (empty)

48 self.producer = producer (empty)

49 

50 def __bool__(self) -> bool: (empty)

51 return True 

52 

53 async def stop(self) -> None: (empty)

54 await self.producer.stop() (empty)

55 

56 @property (empty)

57 def closed(self) -> bool: (empty)

58 return self.producer._closed or False 

59 

60 async def flush(self) -> None: (empty)

61 await self.producer.flush()