Coverage for faststream / confluent / schemas / partition.py: 81%

1from typing import TYPE_CHECKING (empty)

2 

3from confluent_kafka import TopicPartition as ConfluentPartition (empty)

4 

5if TYPE_CHECKING: (empty)

6 from typing_extensions import NotRequired, TypedDict 

7 

8 class _TopicKwargs(TypedDict): 

9 topic: str 

10 partition: int 

11 offset: int 

12 metadata: NotRequired[str] 

13 leader_epoch: NotRequired[int] 

14 

15 

16class TopicPartition: (empty)

17 __slots__ = ( (empty)

18 "leader_epoch", 

19 "metadata", 

20 "offset", 

21 "partition", 

22 "topic", 

23 ) 

24 

25 def __init__( (empty)

26 self, 

27 topic: str, 

28 partition: int = -1, 

29 offset: int = -1001, 

30 metadata: str | None = None, 

31 leader_epoch: int | None = None, 

32 ) -> None: 

33 self.topic = topic (empty)

34 self.partition = partition (empty)

35 self.offset = offset (empty)

36 self.metadata = metadata (empty)

37 self.leader_epoch = leader_epoch (empty)

38 

39 def to_confluent(self) -> ConfluentPartition: (empty)

40 kwargs: _TopicKwargs = { (empty)

41 "topic": self.topic, 

42 "partition": self.partition, 

43 "offset": self.offset, 

44 } 

45 if self.metadata is not None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true(empty)

46 kwargs["metadata"] = self.metadata 

47 if self.leader_epoch is not None: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true(empty)

48 kwargs["leader_epoch"] = self.leader_epoch 

49 return ConfluentPartition(**kwargs) (empty)

50 

51 def add_prefix(self, prefix: str) -> "TopicPartition": (empty)

52 return TopicPartition( (empty)

53 topic=f"{prefix}{self.topic}", 

54 partition=self.partition, 

55 offset=self.offset, 

56 metadata=self.metadata, 

57 leader_epoch=self.leader_epoch, 

58 )