Python中验证Kubernetes容器CPU与内存资源输入字符串的方案咨询

嗨，你的需求我完全get到了——要在API层面提前校验K8s容器资源的输入字符串，避免后续部署踩坑，还要给请求者设置资源上限对吧？刚好有几个靠谱的Python工具能帮你搞定这个事儿，而且都是贴合K8s原生逻辑的，绝对不会有兼容性问题：

1. 首选方案：Kubernetes官方Python客户端的parse_quantity函数 #

这是最推荐的选择，因为它用的就是K8s自身的资源量解析逻辑，准确度拉满，不用担心解析出来的结果和K8s实际识别的不一致。

具体用法： #

先安装依赖：

pip install kubernetes

然后就能写校验逻辑了，示例代码如下：

from kubernetes.utils import parse_quantity
from kubernetes.client.exceptions import ApiException

def validate_resource_input(resource_str, max_threshold, resource_type):
    try:
        # 自动解析为最小单位数值：CPU转成纳核（1核=10^9纳核），内存转成字节（1Gi=2^30字节）
        parsed_value = parse_quantity(resource_str)
        # 和预设阈值直接做数值比较
        if parsed_value > max_threshold:
            raise ValueError(f"输入的{resource_type}超出上限啦！当前值：{resource_str}，最大值：{max_threshold}")
        return parsed_value
    except (ValueError, ApiException) as e:
        raise ValueError(f"无效的{resource_type}输入：{resource_str}，错误原因：{str(e)}")

# 举个例子，设置你的资源上限
MAX_ALLOWED_CPU = 1000000000  # 对应1核CPU
MAX_ALLOWED_MEM = 4 * 1024**3  # 对应4Gi内存

# 测试合法输入
print(validate_resource_input("250m", MAX_ALLOWED_CPU, "CPU"))  # 输出250000000，符合要求
print(validate_resource_input("512M", MAX_ALLOWED_MEM, "内存"))  # 输出536870912，符合要求

# 要是输入格式错了或者超上限，会直接抛出错误提示
# validate_resource_input("2.5X", MAX_ALLOWED_CPU, "CPU")  # 格式非法，触发报错
# validate_resource_input("5Gi", MAX_ALLOWED_MEM, "内存")  # 超出4Gi上限，触发报错

这个函数能自动处理你提到的所有输入格式，不管是4Gi/8G这种内存单位，还是1/250m/0.5这种CPU单位，都能准确解析。

2. 轻量替代：k8s-quantity库 #

如果你觉得整个K8s客户端包太大，只想用解析功能的话，可以试试这个轻量库，它的解析逻辑和官方完全一致，但体积小很多。

具体用法： #

先安装：

pip install k8s-quantity

代码逻辑和上面几乎一样，只是导入的库不同：

from k8s_quantity import parse_quantity

def validate_resource_input(resource_str, max_threshold, resource_type):
    try:
        parsed_value = parse_quantity(resource_str)
        if parsed_value > max_threshold:
            raise ValueError(f"输入的{resource_type}超出上限啦！")
        return parsed_value
    except ValueError as e:
        raise ValueError(f"无效的{resource_type}输入：{resource_str}，错误原因：{str(e)}")

# 阈值设置和测试逻辑和之前完全相同，这里就不重复啦

3. 进阶玩法：结合Pydantic做API请求体校验 #

如果你的API是用Pydantic做请求参数校验的，还能把解析和校验逻辑封装成自定义字段，让整个API的输入校验更优雅：

from pydantic import BaseModel, field_validator
from kubernetes.utils import parse_quantity

class K8sResourceRequest(BaseModel):
    cpu_request: str
    cpu_limit: str
    memory_request: str
    memory_limit: str

    @field_validator("cpu_request", "cpu_limit")
    def check_cpu_limit(cls, value):
        max_cpu = 1000000000  # 1核上限
        try:
            parsed_cpu = parse_quantity(value)
            if parsed_cpu > max_cpu:
                raise ValueError(f"CPU值不能超过1核，当前输入：{value}")
            return value
        except Exception as e:
            raise ValueError(f"CPU输入格式错误：{value}，原因：{str(e)}")

    @field_validator("memory_request", "memory_limit")
    def check_memory_limit(cls, value):
        max_mem = 4 * 1024**3  # 4Gi上限
        try:
            parsed_mem = parse_quantity(value)
            if parsed_mem > max_mem:
                raise ValueError(f"内存值不能超过4Gi，当前输入：{value}")
            return value
        except Exception as e:
            raise ValueError(f"内存输入格式错误：{value}，原因：{str(e)}")

# 测试合法请求
valid_req = K8sResourceRequest(
    cpu_request="250m",
    cpu_limit="1",
    memory_request="512M",
    memory_limit="4Gi"
)

# 测试非法请求会直接抛出校验错误
# invalid_req = K8sResourceRequest(
#     cpu_request="2",  # 超过1核上限
#     cpu_limit="1",
#     memory_request="5Gi",  # 超过4Gi上限
#     memory_limit="4Gi"
# )

用这些方案，你就能在API请求的早期阶段就把非法输入、超上限输入拦下来，完全避免后续K8s部署失败的问题，而且逻辑清晰，维护起来也方便。

备注：内容来源于stack exchange，提问作者Mukund Jalan