Skip to content

AdminService

faststream.confluent.helpers.admin.AdminService #

AdminService()
Source code in faststream/confluent/helpers/admin.py 
17
18
def __init__(self) -> None:
    self.admin_client: AdminClient | None = None

admin_client instance-attribute #

admin_client = None

client property #

client

connect async #

connect(config)
Source code in faststream/confluent/helpers/admin.py 
20
21
22
async def connect(self, config: "ConfluentFastConfig") -> None:
    if self.admin_client is None:
        self.admin_client = AdminClient(config.admin_config)

disconnect async #

disconnect()
Source code in faststream/confluent/helpers/admin.py 
24
25
async def disconnect(self) -> None:
    self.admin_client = None

create_topics #

create_topics(topics)
Source code in faststream/confluent/helpers/admin.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def create_topics(self, topics: list[str]) -> list[CreateResult]:
    create_result = self.client.create_topics(
        [NewTopic(topic, num_partitions=1, replication_factor=1) for topic in topics],
    )

    final_results = []
    for topic, f in create_result.items():
        try:
            f.result()

        except Exception as e:
            if "TOPIC_ALREADY_EXISTS" not in str(e):
                result = CreateResult(topic, e)
            else:
                result = CreateResult(topic, None)

        else:
            result = CreateResult(topic, None)

        final_results.append(result)

    return final_results