- 后端
- 云原生
- 容器编排
【免费下载链接】python
Official Python client library for kubernetes
V1IPBlock 是官方 Kubernetes Python 客户端(含 asyncio 变体)中用于表达NetworkPolicy 规则里 IP 地址段(CIDR)放行与排除的核心数据模型:cidr声明允许访问的网段,except声明在cidr范围内仍需排除的网段。本文以 doc/source/kubernetes.aio.client.models.v1_ip_block.rst 对应的异步(aio)模型为主线,逐字段拆解其属性约束、Pydantic 校验行为、JSON/字典双向序列化接口,并结合V1NetworkPolicyPeer给出构造 NetworkPolicy 的完整可运行示例。读完本文,你将掌握V1IPBlock的全部公开 API、输入别名规则与序列化细节,能够准确地在 asyncio 程序中创建和解析基于 IP 网段的网络策略。
一、模型定位:NetworkPolicy 中的 IP 网段描述
在 Kubernetes 中,NetworkPolicy 通过podSelector圈定策略作用的 Pod 集合,再通过 ingress/egress 规则控制进出流量。V1IPBlock正是这些规则中描述"放行哪一个 IP 网段"的类型。其官方文档语义(见 V1IPBlock.md)为:
IPBlock describes a particular CIDR (Ex. "192.168.1.0/24","2001:db8::/64") that is allowed to the pods matched by a NetworkPolicySpec's podSelector. The except entry describes CIDRs that should not be included within this rule.
即:cidr是允许访问(或允许被访问)的地址块,except列出该地址块内需要排除的子网。两个字段均为 CIDR 字符串,同时支持 IPv4(如192.168.1.0/24)与 IPv6(如2001:db8::/64)。
在 OpenAPI 规格文件 scripts/swagger.json 的v1.IPBlock定义中(第 15018–15037 行),可以确认该模型的结构:
"v1.IPBlock": { "description": "IPBlock describes a particular CIDR ... ", "properties": { "cidr": { "description": "...", "type": "string" }, "except": { "description": "... Except values will be rejected if they are outside the cidr range", "items": { "type": "string" }, "type": "array", "x-kubernetes-list-type": "atomic" } }, "required": ["cidr"], "type": "object" }要点:
cidr是必填字段;except是字符串数组(切片),可选,且标注了x-kubernetes-list-type: atomic,表示该列表在服务端按整体原子替换处理;- 若
except中的 CIDR 超出了cidr的地址范围,请求会被拒绝。
二、字段详解与属性约束
V1IPBlock只包含两个属性,完整参数表如下(字段名、类型与说明均以 v1_ip_block.py 为准):
| 属性名(Python) | 线协议字段(JSON) | 类型 | 必填 | 说明 |
|---|---|---|---|---|
cidr | cidr | StrictStr(str) | 是 | 表示 IPBlock 的 CIDR 字符串,合法示例"192.168.1.0/24"、"2001:db8::/64" |
var_except | except | Optional[List[StrictStr]] | 否 | 不应包含在 IPBlock 内的 CIDR 切片,合法示例同上;超出cidr范围的取值会被拒绝,默认None |
源码中的字段声明如下(v1_ip_block.py):
cidr: StrictStr = Field( description='cidr is a string representing the IPBlock ' 'Valid examples are "192.168.1.0/24" or "2001:db8::/64"' ) var_except: Optional[List[StrictStr]] = Field( default=None, validation_alias=AliasChoices("except", "_except"), serialization_alias="except", description="except is a slice of CIDRs that should not be included " "within an IPBlock ... Except values will be rejected if " "they are outside the cidr range", alias="_except", )三个值得注意的细节:
- 字段名避开 Python 关键字:
except是 Python 保留字,无法作为属性名,因此内部字段被命名为var_except。但通过validation_alias=AliasChoices("except", "_except")与serialization_alias="except",对外 JSON 始终使用except键。 cidr使用StrictStr:Pydantic 严格字符串校验,传入非字符串类型不会被宽松转换,从而保证序列化给 API Server 的 CIDR 始终是字符串。- 模型级校验配置(v1_ip_block.py):
model_config = ConfigDict( validate_by_name=True, # 允许按 Python 字段名传参 validate_by_alias=True, # 也允许按别名(except/_except)传参 validate_assignment=True, # 赋值时同样触发校验 extra="forbid", # 禁止未知字段 protected_namespaces=(), )extra="forbid"意味着传入任何既非cidr、也非except/var_except的键都会直接报错,有助于在编写策略代码时尽早发现拼写错误。
三、输入别名预处理:except与_except的兼容逻辑
由于历史原因,该模型同时接受except与_except两个线协议键。__preprocess_input_names类方法(v1_ip_block.py)在模型校验前完成归一化:
if "except" in obj and "_except" in obj: raise ValueError("%s received both %r and %r" % (cls.__name__, "except", "_except")) if "except" not in obj and "_except" in obj: obj["except"] = obj["_except"] obj.pop("_except", None)规则总结:
- 同时给出
except与_except会抛出ValueError; - 只给
_except时自动映射为except; - 最终以
except作为唯一输入键进入模型校验。
这意味着从from_dict反序列化时,两种键写法都能被接受,但应避免二者同时出现。该处理同时服务于从 JSON 反序列化与从 Python dict 构造两条路径。
四、序列化与反序列化 API 全览
V1IPBlock继承pydantic.BaseModel,并额外实现了 OpenAPI Generator 生成模型标准的一组方法(v1_ip_block.py):
| 方法 | 作用 |
|---|---|
to_str() | 返回pprint.pformat(self.to_dict())的可读字符串表示,__repr__亦复用之 |
to_json() | 返回按别名(cidr、except)序列化的 JSON 字符串 |
from_json(json_str) | 从 JSON 字符串创建V1IPBlock实例 |
to_dict(serialize=False) | 返回字典:serialize=False时键为cidr/_except(公开名),serialize=True时键为cidr/except(线协议名) |
from_dict(obj) | 从字典创建实例,内部先走__preprocess_input_names归一化,再调用model_validate |
__eq__/__ne__ | 基于to_dict()的结果比较两个模型是否相等 |
其中to_dict的两种输出模式对应生成器的"legacy / modern"兼容设计:默认返回 Python 公开名(_except),需要交给 API 客户端时可用serialize=True得到线协议名(except)。from_json是from_dict(json.loads(json_str))的便捷封装。
一个完整的构造与往返示例(可直接复制运行):
from kubernetes.aio.client.models.v1_ip_block import V1IPBlock # 方式一:按字段名直接构造(except 是关键字,需用 var_except) block = V1IPBlock( cidr="192.168.1.0/24", var_except=["192.168.1.10/32", "192.168.1.11/32"], ) # 方式二:按线协议键通过 dict 构造 block2 = V1IPBlock.from_dict({ "cidr": "2001:db8::/64", "except": ["2001:db8::1/128"], # 或 "_except": [...] }) print(block) # to_str() -> pprint 输出 print(block.to_json()) # {"cidr": "192.168.1.0/24", "except": [...]} d = block.to_dict(serialize=True) # {'cidr': ..., 'except': [...]} restored = V1IPBlock.from_dict(d) assert block == restored # 同时传入 except 与 _except 会抛 ValueError # V1IPBlock.from_dict({"cidr": "10.0.0.0/8", "except": [], "_except": []})五、在 NetworkPolicy 中的实际使用
V1IPBlock本身并不单独出现在策略 YAML 顶层,而是作为peer(对端)的一个可选项被引用。查看 v1_network_policy_peer.py 可见:
class V1NetworkPolicyPeer(BaseModel): ip_block: Optional[V1IPBlock] = Field( default=None, validation_alias=AliasChoices("ipBlock", "ip_block"), serialization_alias="ipBlock", ) namespace_selector: Optional[V1LabelSelector] = ... pod_selector: Optional[V1LabelSelector] = ...OpenAPI 对v1.NetworkPolicyPeer的约束(scripts/swagger.json)明确指出:
If this field [ipBlock] is set then neither of the other fields can be.
也就是说,V1NetworkPolicyPeer中ipBlock、namespaceSelector、podSelector三选一,ipBlock一旦设置,其余两个选择器字段必须为空。V1NetworkPolicyPeer随后被V1NetworkPolicyIngressRule(from列表)与V1NetworkPolicyEgressRule(to列表)引用,形成完整的 ingress/egress 规则。
下面是一个在 asyncio 客户端中构造"允许从192.168.1.0/24访问,但排除其中两个管理地址"的 NetworkPolicy 示例:
import asyncio from kubernetes import config from kubernetes.aio import client as aio_client from kubernetes.aio.client.models import ( V1IPBlock, V1NetworkPolicy, V1NetworkPolicyPeer, V1NetworkPolicyIngressRule, V1NetworkPolicySpec, ) async def create_ip_policy(): await config.load_kube_config() # 或用 load_incluster_config() async with aio_client.ApiClient() as api_client: api = aio_client.NetworkingV1Api(api_client) peer = V1NetworkPolicyPeer( ip_block=V1IPBlock( cidr="192.168.1.0/24", var_except=["192.168.1.10/32", "192.168.1.11/32"], ) ) policy = V1NetworkPolicy( api_version="networking.k8s.io/v1", kind="NetworkPolicy", metadata={"name": "allow-app-cidr", "namespace": "default"}, spec=V1NetworkPolicySpec( pod_selector={"matchLabels": {"app": "my-app"}}, ingress=[V1NetworkPolicyIngressRule(from_=[peer])], policy_types=["Ingress"], ), ) await api.create_namespaced_network_policy( namespace="default", body=policy ) # 读取回查,确认服务端接受的 IPBlock got = await api.read_namespaced_network_policy( name="allow-app-cidr", namespace="default" ) print(got.spec.ingress[0].from_[0].ip_block.to_dict(serialize=True)) asyncio.run(create_ip_policy())等价于以下 Kubernetes YAML 语义:
apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: allow-app-cidr namespace: default spec: podSelector: matchLabels: app: my-app policyTypes: ["Ingress"] ingress: - from: - ipBlock: cidr: 192.168.1.0/24 except: - 192.168.1.10/32 - 192.168.1.11/32六、同步与异步客户端的模型一致性
仓库同时维护两个版本的V1IPBlock:
- 异步(asyncio):kubernetes/aio/client/models/v1_ip_block.py
- 同步:kubernetes/client/models/v1_ip_block.py
两者内容完全一致(均基于 OpenAPI release-1.37 规格生成),字段、别名、校验逻辑与方法签名完全相同。差异仅体现在上层 API 客户端的使用方式:同步客户端直接调用kubernetes.client.NetworkingV1Api,而 asyncio 客户端使用kubernetes.aio.client.NetworkingV1Api并配合async with与await。因此本文所有关于V1IPBlock的字段与序列化结论,对两种模式同样适用;选择哪个包,取决于你的应用是否基于事件循环。
七、常见陷阱与最佳实践
except必须落在cidr范围内:这是 Kubernetes API Server 的硬性校验。例如cidr: 10.0.0.0/8搭配except: ["172.16.0.0/16"]会被拒绝,应在本地编码时就校验(可借助ipaddress模块):import ipaddress cidr = ipaddress.ip_network("192.168.1.0/24") for exc in ["192.168.1.10/32"]: assert ipaddress.ip_network(exc).subnet_of(cidr), f"{exc} 超出 {cidr}"- 同时支持 IPv4 与 IPv6:
cidr与except类型均为普通字符串,客户端不做 CIDR 语法校验(校验由 API Server 完成),但示例写法应遵循标准 CIDR 格式,如2001:db8::/64。 - 不要同时传
except与_except:__preprocess_input_names会直接抛ValueError。推荐统一使用except键(或 Python 侧构造时用var_except=)。 extra="forbid"会拒绝未知字段:构造时传入cidr、except以外的键(如笔误的cidrs)会立即报错,属于客户端提供的第一道防线。- 优先使用
from_dict处理来自 API 的响应:读取 NetworkPolicy 后拿到的响应 dict 可通过V1IPBlock.from_dict(...)/from_json(...)还原为类型化对象,便于后续比较(==基于to_dict()逐字段比较)。
八、相关资源
- 文档入口:doc/source/kubernetes.aio.client.models.v1_ip_block.rst
- 异步模型实现:kubernetes/aio/client/models/v1_ip_block.py
- 同步模型实现:kubernetes/client/models/v1_ip_block.py
- 模型速查文档:kubernetes/aio/docs/V1IPBlock.md
- 引用方
V1NetworkPolicyPeer:kubernetes/client/models/v1_network_policy_peer.py - OpenAPI 原始定义(
v1.IPBlock):scripts/swagger.json
- 后端
- 云原生
- 容器编排
【免费下载链接】python
Official Python client library for kubernetes
相关推荐
Kubernetes Python 客户端 V1CustomResourceDefinitionNames 模型详解:CRD 命名配置的字段、别名与实战用法
Kubernetes Python 客户端 V1CustomResourceDefinitionNames 模型详解:CRD 命名配置的字段、别名与实战用法 导
后端云原生容器编排Kubernetes Python 客户端(aio)V1StorageOSVolumeSource 模型详解:StorageOS 卷的字段、别名映射与序列化机制
Kubernetes Python 客户端(aio)V1StorageOSVolumeSource 模型详解:StorageOS 卷的字段、别名映射与序列化机制
后端云原生容器编排python-kubernetes aio 客户端中 V1StatefulSet 模型详解:字段、序列化与异步用法
python kubernetes aio 客户端中 V1StatefulSet 模型详解:字段、序列化与异步用法 V1StatefulSet 是 kubern
后端云原生容器编排
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考