Coverage for faststream / confluent / helpers / admin.py: 91%

1from dataclasses import dataclass (empty)

2from typing import TYPE_CHECKING (empty)

3 

4from confluent_kafka.admin import ( # type: ignore[attr-defined] (empty)

5 AdminClient, 

6 NewTopic, 

7) 

8 

9if TYPE_CHECKING: (empty)

10 from .config import ConfluentFastConfig 

11 

12 

13@dataclass (empty)

14class CreateResult: (empty)

15 topic: str (empty)

16 error: Exception | None (empty)

17 

18 

19class AdminService: (empty)

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

21 self.admin_client: AdminClient | None = None (empty)

22 

23 async def connect(self, config: "ConfluentFastConfig") -> None: (empty)

24 if self.admin_client is None: 24 ↛ exitline 24 didn't return from function 'connect' because the condition on line 24 was always true(empty)

25 self.admin_client = AdminClient(config.admin_config) (empty)

26 

27 async def disconnect(self) -> None: (empty)

28 self.admin_client = None (empty)

29 

30 @property (empty)

31 def client(self) -> AdminClient: (empty)

32 assert self.admin_client is not None, ( (empty)

33 "Admin client was not connected. Please, connect the broker first." 

34 ) 

35 return self.admin_client (empty)

36 

37 def create_topics(self, topics: list[str]) -> list[CreateResult]: (empty)

38 create_result = self.client.create_topics( (empty)

39 [NewTopic(topic, num_partitions=1, replication_factor=1) for topic in topics], 

40 ) 

41 

42 final_results = [] (empty)

43 for topic, f in create_result.items(): (empty)

44 try: (empty)

45 f.result() (empty)

46 

47 except Exception as e: (empty)

48 if "TOPIC_ALREADY_EXISTS" not in str(e): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true(empty)

49 result = CreateResult(topic, e) 

50 else: 

51 result = CreateResult(topic, None) (empty)

52 

53 else: 

54 result = CreateResult(topic, None) (empty)

55 

56 final_results.append(result) (empty)

57 

58 return final_results (empty)