使用 kr8s 库将 Kubernetes 资源导出为 YAML 清单

pip install kr8s pyyaml

import yaml
import kr8s

async def dump_kubernetes_resource_to_yaml(resource_name: str, namespace: str, resource_type: str = "Deployment"):
    """
    从 Kubernetes 集群中获取指定资源，并将其导出为 YAML 格式。

    Args:
        resource_name (str): 要导出的资源名称。
        namespace (str): 资源所在的命名空间。
        resource_type (str): 资源的类型，例如 "Deployment", "Service", "Pod" 等。
    """
    try:
        # 初始化 kr8s 客户端
        # kr8s 会自动查找 kubeconfig 文件或使用集群内配置
        api = await kr8s.api()

        # 根据资源类型动态获取对应的类
        # 注意：kr8s.objects 提供了所有常见资源类型的直接访问
        # 例如 kr8s.objects.Deployment, kr8s.objects.Service 等
        resource_class = getattr(kr8s.objects, resource_type)

        # 获取指定名称和命名空间的资源对象
        resource_obj = await resource_class.get(resource_name, namespace=namespace)

        if resource_obj:
            # 调用 to_dict() 方法获取资源的字典表示
            resource_dict = resource_obj.to_dict()

            # 使用 pyyaml 将字典序列化为 YAML 字符串
            # safe_dump() 是推荐的序列化方法
            yaml_output = yaml.safe_dump(resource_dict, default_flow_style=False, sort_keys=False)

            print(f"--- {resource_type}/{resource_name} (Namespace: {namespace}) ---")
            print(yaml_output)
            print("-" * 50)
        else:
            print(f"错误: 未找到 {namespace} 命名空间下的 {resource_type} 资源 '{resource_name}'。")

    except kr8s.exceptions.NotFoundError:
        print(f"错误: 未找到 {namespace} 命名空间下的 {resource_type} 资源 '{resource_name}'。")
    except Exception as e:
        print(f"发生未知错误: {e}")

# 运行异步函数
if __name__ == "__main__":
    import asyncio
    # 替换为您的实际资源名称、命名空间和类型
    asyncio.run(dump_kubernetes_resource_to_yaml("example", "example", "Deployment"))
    # 您也可以尝试导出其他资源，例如：
    # asyncio.run(dump_kubernetes_resource_to_yaml("my-service", "default", "Service"))

--- Deployment/example (Namespace: example) ---
apiVersion: apps/v1
kind: Deployment
metadata:
  annotations:
    deployment.kubernetes.io/revision: '1'
  creationTimestamp: '2023-12-04T16:08:14Z'
  generation: 2
  labels:
    app: example
  name: example
  namespace: example
  resourceVersion: '2570142'
  uid: 0555bacf-94a7-43b3-8b89-e20573bdb256
spec:
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: example
  strategy:
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
    type: RollingUpdate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: example
    spec:
      containers:
      - image: ghcr.io/larsks/example:42843ba7
        imagePullPolicy: IfNotPresent
        name: example
        ports:
        - containerPort: 3141
          protocol: TCP
        resources: {}
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        volumeMounts:
        - mountPath: /data
          name: example-data
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30
      volumes:
      - name: example-data
        persistentVolumeClaim:
          claimName: example-data
status:
  # ... (status字段通常包含运行时信息，通常在导出用于重新应用的清单时会被移除)
--------------------------------------------------