Skip to content

AdminService

faststream.confluent.helpers.admin.AdminService #

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

admin_client instance-attribute #

admin_client: AdminClient | None = None

client property #

client: AdminClient

connect async #

connect(
    config: ConfluentFastConfig,
    logger: LoggerState | None = None,
) -> None
Source code in faststream/confluent/helpers/admin.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
async def connect(
    self,
    config: "ConfluentFastConfig",
    logger: "LoggerState | None" = None,
) -> None:
    if self.admin_client is None:
        admin_config = config.admin_config
        if logger is not None:
            self.admin_client = AdminClient(
                admin_config,
                logger=_LazyLoggerProxy(logger),
            )
        else:
            self.admin_client = AdminClient(admin_config)

disconnect async #

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

create_topics #

create_topics(topics: list[str]) -> list[CreateResult]
Source code in faststream/confluent/helpers/admin.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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