Imports

To use Confluent Admin API SDK in a project

from confluent_cloud_sdk.client_factory import ConfluentClient
from confluent_cloud_sdk.confluent_iam_v2 import ApiKey
from confluent_cloud_sdk.confluent_iam_v2 import ServiceAccount

Initialize connection

client = ConfluentClient(
    "cloud_key_key",
    "cloud_key_secret",
)

Example with secret in AWS And list all assets

import json
from os import environ

from boto3.session import Session
from confluent_cloud_sdk.client_factory import ConfluentClient
from confluent_cloud_sdk.confluent_iam_v2 import ServiceAccount
from confluent_cloud_sdk.confluent_org_v2 import ConfluentEnvironment
from confluent_cloud_sdk.confluent_cluster_v2 import KafkaClusterV2

from compose_x_common.aws import get_session


def get_confluent_admin_secret(
    secret_arn: str,
    session: Session = None,
    key_id: str = "ApiKey",
    secret_id: str = "ApiSecret",
) -> ConfluentClient:
    session = get_session(session)
    client = session.client("secretsmanager")
    value = json.loads(client.get_secret_value(SecretId=secret_arn)["SecretString"])
    return ConfluentClient(value[key_id], value[secret_id])


cclient = get_confluent_admin_secret(environ.get("SECRET_ARN"))

envs = cclient.list_all(ConfluentEnvironment)

for env in envs:
    print(env.obj_id)
    clusters = cclient.list_all(KafkaClusterV2, url_args=f"?environment={env.obj_id}")
    for cluster in clusters:
        print(cluster.obj_id)

svc_accounts = cclient.list_all(ServiceAccount)
for svc_account in svc_accounts:
    print(svc_account.obj_id)
    svc_account.import_api_keys()
    print([key.obj_id for key in svc_account.api_keys.values()])