如何用PyYAML提取指定YAML节点生成OpenShift/K8s部署配置文件？

前置准备 #

先安装依赖库：

pip install pyyaml

核心处理逻辑 #

以下是完整的Python脚本，实现从已有资源YAML中提取并生成极简可创建的YAML文件：

import yaml
import os
from typing import Dict, List

# -------------------------- 通用工具函数 --------------------------
def clean_metadata(metadata: Dict) -> Dict:
    """清理metadata，只保留手动配置字段，移除自动生成内容"""
    keep_fields = ["name", "namespace", "labels", "annotations"]
    cleaned = {k: v for k, v in metadata.items() if k in keep_fields and v is not None}
    return cleaned

def save_to_yaml(data: List[Dict], output_path: str):
    """将多个资源YAML保存为多文档文件"""
    with open(output_path, "w") as f:
        yaml.dump_all(data, f, sort_keys=False, default_flow_style=False)

# -------------------------- PVC 处理 --------------------------
def extract_pvc_names_from_deployment(deployment_yaml_path: str) -> List[str]:
    """从Deployment YAML中提取关联的PVC名称"""
    with open(deployment_yaml_path, "r") as f:
        dep_data = yaml.safe_load(f)
    
    pvc_names = []
    if "spec" in dep_data and "template" in dep_data["spec"]:
        volumes = dep_data["spec"]["template"]["spec"].get("volumes", [])
        for vol in volumes:
            if "persistentVolumeClaim" in vol:
                pvc_names.append(vol["persistentVolumeClaim"]["claimName"])
    return list(set(pvc_names))  # 去重

def clean_pvc_yaml(pvc_yaml_path: str) -> Dict:
    """清理PVC YAML，生成可创建的极简版本"""
    with open(pvc_yaml_path, "r") as f:
        pvc_data = yaml.safe_load(f)
    
    cleaned_pvc = {
        "apiVersion": pvc_data["apiVersion"],
        "kind": pvc_data["kind"],
        "metadata": clean_metadata(pvc_data["metadata"]),
        "spec": {
            "accessModes": pvc_data["spec"]["accessModes"],
            "resources": pvc_data["spec"]["resources"],
            "storageClassName": pvc_data["spec"].get("storageClassName")
        }
    }
    # 移除值为None的字段
    cleaned_pvc["spec"] = {k: v for k, v in cleaned_pvc["spec"].items() if v is not None}
    return cleaned_pvc

# -------------------------- Service 处理 --------------------------
def clean_service_yaml(service_yaml_path: str) -> Dict:
    """清理Service YAML，生成可创建的极简版本"""
    with open(service_yaml_path, "r") as f:
        svc_data = yaml.safe_load(f)
    
    cleaned_svc = {
        "apiVersion": svc_data["apiVersion"],
        "kind": svc_data["kind"],
        "metadata": clean_metadata(svc_data["metadata"]),
        "spec": {
            "selector": svc_data["spec"]["selector"],
            "ports": svc_data["spec"]["ports"],
            "type": svc_data["spec"].get("type", "ClusterIP")
        }
    }
    return cleaned_svc

# -------------------------- Deployment 处理 --------------------------
def clean_deployment_yaml(deployment_yaml_path: str) -> Dict:
    """清理Deployment YAML，生成可创建的极简版本"""
    with open(deployment_yaml_path, "r") as f:
        dep_data = yaml.safe_load(f)
    
    # 清理Pod模板的metadata
    pod_template_metadata = clean_metadata(dep_data["spec"]["template"]["metadata"])
    
    cleaned_dep = {
        "apiVersion": dep_data["apiVersion"],
        "kind": dep_data["kind"],
        "metadata": clean_metadata(dep_data["metadata"]),
        "spec": {
            "replicas": dep_data["spec"].get("replicas", 1),
            "selector": dep_data["spec"]["selector"],
            "template": {
                "metadata": pod_template_metadata,
                "spec": dep_data["spec"]["template"]["spec"]
            }
        }
    }
    
    # 移除status字段（如果存在）
    if "status" in cleaned_dep:
        del cleaned_dep["status"]
    
    return cleaned_dep

# -------------------------- 主流程 --------------------------
def main():
    # 替换为你的实际文件路径
    deployment_yaml = "target_deployment.yaml"
    service_yaml = "target_service.yaml"
    output_file = "new_deployment_resources.yaml"
    
    # 提取PVC名称
    pvc_names = extract_pvc_names_from_deployment(deployment_yaml)
    print(f"提取到PVC列表: {pvc_names}")
    
    # 清理每个PVC的YAML
    cleaned_pvcs = []
    for name in pvc_names:
        pvc_path = f"pvc-{name}.yaml"
        if os.path.exists(pvc_path):
            cleaned_pvcs.append(clean_pvc_yaml(pvc_path))
        else:
            print(f"警告: 未找到PVC文件 {pvc_path}")
    
    # 清理Service YAML
    cleaned_svc = clean_service_yaml(service_yaml)
    
    # 清理Deployment YAML
    cleaned_dep = clean_deployment_yaml(deployment_yaml)
    
    # 合并所有资源并保存
    all_resources = cleaned_pvcs + [cleaned_svc, cleaned_dep]
    save_to_yaml(all_resources, output_file)
    print(f"已生成极简创建YAML: {output_file}")

if __name__ == "__main__":
    main()

使用说明 #

1. 先用oc/kubectl导出所需资源的YAML：
   • 导出Deployment：oc get deployment <deploy-name> -o yaml > target_deployment.yaml
   • 导出关联Service：oc get service <service-name> -o yaml > target_service.yaml
   • 导出每个PVC：oc get pvc <pvc-name> -o yaml > pvc-<pvc-name>.yaml
2. 修改脚本中的文件路径为你的实际路径
3. 运行脚本，生成的new_deployment_resources.yaml即可直接用于创建新资源

关键说明 #

• clean_metadata函数仅保留name、namespace、labels、annotations等手动配置字段，自动生成的uid、resourceVersion等会被移除
• 各资源的spec仅保留核心创建所需字段：PVC保留存储配置、Service保留路由规则、Deployment保留副本数与Pod模板
• 脚本会自动对PVC列表去重，避免重复生成资源配置

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