如何使用Python执行ccloud命令创建Topic

在Python中执行外部ccloud命令推荐使用标准库的subprocess模块，相比废弃的os.system，它可以更好地控制执行流程、捕获输入输出、处理异常，同时支持更安全的参数传递方式。

场景1：基础执行（仅完成创建，不需要处理命令输出） #

优先使用列表形式传参，避免shell注入风险：

import subprocess

try:
    # check=True参数会在命令执行失败时主动抛出异常，方便捕获处理
    subprocess.run(["ccloud", "create", "topic", "mytopic"], check=True)
    print("Topic创建成功")
except subprocess.CalledProcessError as e:
    print(f"Topic创建失败，错误码：{e.returncode}")
except FileNotFoundError:
    print("未找到ccloud命令，请确认已安装ccloud CLI且已添加到系统环境变量PATH中")

如果需要额外传参（比如指定集群ID、分区数、副本数），直接把参数追加到列表中即可，示例：["ccloud", "create", "topic", "mytopic", "--cluster", "lkc-xxxx", "--partitions", "6"]

场景2：需要捕获命令的输出和错误信息 #

如果需要拿到命令执行结果用于后续逻辑，可以添加输出捕获配置：

import subprocess

try:
    result = subprocess.run(
        ["ccloud", "create", "topic", "mytopic"],
        check=True,
        capture_output=True,
        text=True # 直接返回字符串格式的输出，不需要手动做二进制解码
    )
    print("Topic创建成功，命令返回内容：", result.stdout)
except subprocess.CalledProcessError as e:
    print(f"Topic创建失败，错误码：{e.returncode}")
    print("错误详情：", e.stderr)

免提前登录的配置方式 #

如果运行环境没有提前执行ccloud login做认证，可以通过环境变量传入认证信息：

import subprocess
import os

env = os.environ.copy()
# 按需传入认证参数，也支持CCLOUD_API_KEY、CCLOUD_API_SECRET等鉴权配置
env["CCLOUD_EMAIL"] = "你的ccloud账号邮箱"
env["CCLOUD_PASSWORD"] = "你的ccloud账号密码"

try:
    subprocess.run(
        ["ccloud", "create", "topic", "mytopic"],
        check=True,
        env=env,
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as e:
    print(f"执行失败：{e.stderr}")

内容的提问来源于stack exchange，提问作者Mehmood ulhassan