企业官网建设流程全解析

• 后端
• 云原生
• 容器编排

导读

kubernetes.aio.dynamic.client是 Kubernetes 官方 Python 客户端（当前仓库python）中基于asyncio的**动态客户端（DynamicClient）**模块。与针对每个 API 群组生成静态方法（如CoreV1Api）的客户端不同，动态客户端在运行时向集群 API Server 发起 discovery 请求，动态发现资源类型并以统一的Resource对象执行增删改查。本文面向希望用异步方式操作原生资源与 CRD（Custom Resource Definition）的开发者，读完你将掌握：如何初始化异步动态客户端、如何通过resources.get(...)发现资源、如何使用get/create/delete/replace/patch/server_side_apply/watch等核心方法，以及背后的资源对象模型、异常体系与缓存机制。文中所有代码均可直接复制运行（需可访问的集群与 kubeconfig）。

一、模块定位与适用场景

本模块由文档页 kubernetes.aio.dynamic.client.rst 通过automodule自动生成，其源码位于 kubernetes/aio/dynamic/client.py。它对外导出：

• DynamicClient—— 动态客户端主体；
• Resource/ResourceList/ResourceInstance/ResourceField/Subresource—— 资源对象模型（见 resource.py）；
• EagerDiscoverer/LazyDiscoverer—— 资源发现策略（见 discovery.py）。

适用场景：在开发 Operator、控制器、自定义控制器或管理 CRD 的工具时，静态生成的 API 无法覆盖集群中未知/新增的资源类型，而动态客户端可以在运行时发现并操作任意已注册资源。同步版动态客户端位于 kubernetes/dynamic/，而本模块是其在asyncio体系下的等价实现，依赖kubernetes.aio的异步ApiClient。

二、环境准备与客户端初始化

异步动态客户端需要asyncio环境。当前仓库提供异步专属依赖与安装入口：requirements-asyncio.txt、setup-asyncio.py。运行时需要：

• 一个可访问的 Kubernetes 集群；
• 有效的 kubeconfig（或 in-cluster 配置）；
• 异步 HTTP 客户端（kubernetes.aio.client.ApiClient）。

2.1 标准初始化流程

参考官方示例 configmap.py：

import asyncio from kubernetes.aio.client import api_client from kubernetes.aio.client.configuration import Configuration from kubernetes.aio.config import kube_config from kubernetes.aio.dynamic import DynamicClient async def main(): config = Configuration() await kube_config.load_kube_config(client_configuration=config) async with api_client.ApiClient(configuration=config) as apic: client = await DynamicClient(apic) # ... 使用 client if __name__ == "__main__": loop = asyncio.new_event_loop() loop.run_until_complete(main()) loop.close()

初始化时，DynamicClient.__await__会调用discoverer(self, cache_file)并执行资源发现（见 client.py）：

def __await__(self): async def closure(): self.__discoverer = await self.discoverer(self, self.cache_file) return self return closure().__await__()

因此await DynamicClient(apic)是必须的。也可以使用异步上下文管理器：

async with DynamicClient(apic) as client: ...

其__aenter__与__await__等价地完成 discoverer 初始化（见 client.py），示例 accept_header.py 采用的就是这种写法。

2.2 构造参数

DynamicClient.__init__(client, cache_file=None, discoverer=None)（见 client.py）：

初始化后可通过两个属性访问内部状态：

• client.resources：返回 discoverer 对象，用于搜索资源；
• client.version：返回从/version端点获取的集群版本信息（见 client.py）。

三、资源发现：LazyDiscoverer 与 EagerDiscoverer

动态客户端之所以"动态"，核心在于 discovery 机制。抽象基类Discoverer负责：

• 向集群请求 API 组信息（parse_api_groups）；
• 按(prefix, group, version)拉取每个 API 组的资源清单（get_resources_for_api_version）；
• 将结果构建为Resource/ResourceList对象；
• 写入/读取本地缓存文件（见 discovery.py）。

3.1 两种发现策略

• LazyDiscoverer（默认）：discover()只加载 API 组骨架，不立即请求每个组的资源；当search()命中某个尚未拉取资源的组时，才按需向集群请求该组的资源清单（discovery.py）。适合资源种类多的集群，首启开销小。
• EagerDiscoverer：discover()一次性拉取所有 API 组的全部资源（request_resources=True），适合对首查延迟敏感、资源种类可控的场景（discovery.py）。

3.2 本地缓存机制

Discoverer.__init__会根据client.configuration.host的 MD5 值生成默认缓存文件名osrcp-<md5>.json，存放于系统临时目录（discovery.py）。缓存中记录了library_version，当客户端库版本变化时缓存会被自动判定失效并刷新；search在本地未命中时也会触发invalidate_cache()重新发现，以感知新创建的 CRD。缓存通过CacheEncoder/CacheDecoder以_type字段实现对象与 JSON 的互相转换（discovery.py）。

3.3 查找资源：get 与 search

resources对象提供两个关键方法（见 discovery.py）：

• search(prefix=..., group=..., api_version=..., kind=..., **kwargs)：返回所有匹配的资源对象（列表）；api_version中若包含/（如apps/v1），会被自动拆分为group与api_version。
• get(**kwargs)：基于search的结果做消歧，恰好匹配一个时返回该资源，无匹配抛ResourceNotFoundError，多匹配抛ResourceNotUniqueError。若有多个匹配，优先选择api_version精确匹配者、其次优先非List类型。

最常见的用法是：

# 内置资源 api = await client.resources.get(api_version="v1", kind="ConfigMap") deploy = await client.resources.get(api_version="apps/v1", kind="Deployment") # 自定义资源（CRD） crd_api = await client.resources.get( api_version="apiextensions.k8s.io/v1", kind="CustomResourceDefinition" ) ingressroute_api = await client.resources.get( api_version="apps.example.com/v1", kind="IngressRoute" )

注意：创建 CRD 后，discovery 缓存需要短暂刷新。官方示例与 e2e 测试采用"先捕获ResourceNotFoundError，await asyncio.sleep(2)后重试"的策略（见 namespaced_custom_resource.py）。

四、核心 CRUD 操作

DynamicClient的 CRUD 方法统一签名风格：await api.get(...)、await api.create(body=..., namespace=...)。这些方法本质上都是先构建资源 URL 路径，再调用统一的request方法。

4.1 Resource.path 与 URL 构建

Resource.path(name=None, namespace=None)根据资源的namespaced标志和传入参数，从urls字典中选取合适的模板并格式化（resource.py）：

urls = { 'base': '/{prefix}/{group_version}/{name_lower}', 'namespaced_base': '/{prefix}/{group_version}/namespaces/{namespace}/{name_lower}', 'full': '/{prefix}/{group_version}/{name_lower}/{name}', 'namespaced_full': '/{prefix}/{group_version}/namespaces/{namespace}/{name_lower}/{name}', }

4.2 获取：get

# 读取单个对象 pod = await api.get(name="my-pod", namespace="default") # 列出对象（支持 label/field 选择器） pods = await api.get(namespace="default", label_selector="app=nginx")

get(resource, name=None, namespace=None, **kwargs)将name/namespace交给resource.path()，并把label_selector、field_selector、pretty、limit、_continue、resource_version等透传给底层请求（client.py）。

4.3 创建：create

await api.create(body=configmap_manifest, namespace="default")

create的处理逻辑（client.py）：

1. serialize_body(body)：若body是ResourceInstance（具备to_dict），先转回普通 dict，否则原样使用；
2. 若资源是namespaced的，调用ensure_namespace—— 优先取namespace参数，其次取body['metadata']['namespace']，两者皆无则抛出ValueError；
3. 以POST请求resource.path(namespace=...)。

4.4 删除：delete

await api.delete(name="my-cm", namespace="default") await api.delete(name="my-rc", namespace="default", propagation_policy="Background")

delete的校验逻辑（client.py）值得注意：

• name、label_selector、field_selector至少提供其一，否则抛ValueError("At least one of name|label_selector|field_selector is required")；
• 对 namespaced 资源，还必须提供namespace、label_selector或field_selector之一，否则抛ValueError。

它支持propagation_policy、grace_period_seconds、orphan_dependents、dry_run等查询参数，例如测试中用propagation_policy='Background'删除 ReplicationController（见 client_test.py）。

4.5 整体替换：replace

await api.replace(body=deployment_manifest, name=name, namespace="default")

replace基于PUT语义：name可从参数或body['metadata']['name']推断，缺省抛ValueError；namespaced 资源同样需要 namespace（client.py）。

4.6 局部更新：patch

await api.patch( body=configmap_manifest, name=configmap_name, namespace="default", )

patch默认的Content-Type为application/strategic-merge-patch+json（Kubernetes 内置资源的策略合并补丁）；对 CRD（无策略合并语义）应显式指定content_type="application/merge-patch+json"，示例见 namespaced_custom_resource.py。patch也会从body['metadata']['name']推断name（client.py）。

4.7 服务端应用：server_side_apply

resp = await api.server_side_apply( namespace="default", body=pod_manifest, field_manager="kubernetes-unittests", dry_run="All", )

server_side_apply强制使用Content-Type: application/apply-patch+yaml，并支持force_conflicts（转为查询参数force）、field_manager、dry_run（client.py）。e2e 测试通过检查resp.metadata.managedFields[0].manager验证 field manager 生效（client_test.py）。

五、异步 Watch：实时监听资源事件

DynamicClient.watch是静态方法，用于流式监听资源事件（client.py）：

async for event in client.watch(api, timeout=3, namespace="default", name=name): print(event['type']) # ADDED / MODIFIED / DELETED 等 print(event['object'].metadata)

参数与返回说明（源码 docstring）：

每个事件是包含type、raw_object、object三个键的字典，其中object被包装为ResourceInstance，因此可以用点号访问字段。timeout传None时流不会自行终止——e2e 测试用asyncio.wait_for(..., timeout=5)验证了这一点（client_test.py）。参考用法见 client.py 中的 docstring 示例。

六、底层 request 与查询参数体系

所有高层方法最终汇聚到request(method, path, body=None, **params)（client.py）。它被@meta_request装饰器包裹：装饰器负责把ApiException翻译成dynamic.exceptions中的语义化异常，并把响应 JSON 序列化为ResourceInstance（可通过serialize=False关闭、用serializer=替换序列化器，见 client.py）。

request内部完成：路径补/前缀、查询参数组装、Accept/Content-Type头设置、BearerToken 认证，然后通过param_serialize+call_api发起请求。支持的查询参数（下划线命名 → HTTP 参数名）包括：

头部处理上：Accept默认协商application/json与application/yaml；Content-Type默认application/json（discovery 路由不接受通配*/*）；可通过header_params传入自定义头。测试中通过自定义Accept: application/json;as=PartialObjectMetadataList;v=v1;g=meta.k8s.io获取 PartialObjectMetadata 列表（client_test.py，示例见 accept_header.py）。

请求超时可通过_request_timeout参数控制，示例 request_timeout.py 在每次调用中传入_request_time=60（60 秒客户端超时）。

七、资源对象模型：Resource、ResourceInstance、ResourceField 等

7.1 Resource

Resource代表一种 API 资源类型，保存构建 URL 所需的信息：prefix、group、api_version、kind、namespaced、verbs、name、singular_name、short_names、subresources等。group_version属性在存在 group 时返回group/version（如apps/v1），否则仅返回版本。构造时要求api_version、kind、prefix至少不为空（resource.py）。

Resource.__getattr__有两个巧妙行为：

• 若访问的属性名匹配某个subresource，返回对应的Subresource对象；
• 否则返回partial(getattr(self.client, name), self)，即把DynamicClient的方法（如get、create）绑定到该资源上。这就是await api.create(...)、await api.get(...)能够直接调用、且api自动作为resource参数传入的原因。

7.2 Subresource

Subresource表示资源的子资源（如scale、status），URL 形如/apis/{group}/{version}/namespaces/{ns}/{parent}/{name}/{subresource}，同样继承DynamicClient的 CRUD 方法（resource.py）。

7.3 ResourceList

ResourceList表示资源的*List类型，持有base_kind，支持对 List body 中每个 item 批量执行get、delete、create、replace、patch（verb_mapper）。注意其源码注释标注部分方法"未被任何测试场景执行，是否必要待确认"，批量语义请以实际行为为准（resource.py）。

7.4 ResourceInstance 与 ResourceField

ResourceInstance是把 API 响应解析后的实例，核心价值是支持点号访问：resp.metadata.name、resp.spec.replicas、resp.status.conditions[0]["type"]。它递归地把 dict 解析为ResourceField（__getattr__返回None而非抛错，从而能通过hasattr判断）、把 list 解析为 list；同时保留resp['items']下标访问与resp.items点号访问两种形式。to_dict()可随时还原为纯 dict（resource.py）。__repr__会用 YAML 格式化打印对象，便于调试。

7.5 序列化

serialize_body(body)（client.py）接受dict或ResourceInstance：对具备to_dict的对象调用之，否则原样返回（空值返回{}）。序列化测试覆盖了 dict、ResourceInstance、ResourceField 三种输入（client_test.py）。

八、异常体系：语义化 HTTP 错误

动态客户端将底层ApiException通过api_exception()映射为语义化异常（exceptions.py）：

未列出的状态码映射为通用DynamicApiError。所有动态异常继承ApiException，保留status、reason、body、headers，并额外携带original_traceback；summary()可从 JSON body 中提取message字段。另有非 HTTP 异常：ResourceNotFoundError（资源未发现）、ResourceNotUniqueError（匹配到多个资源）、KubernetesValidateMissing（未安装kubernetes-validate）。

九、资源定义校验：validate

DynamicClient.validate(definition, version=None, strict=False)（client.py）用于校验资源定义是否合法：

• 依赖可选包kubernetes_validate，未安装时抛KubernetesValidateMissing；
• version未指定时，优先取self.version['kubernetes']['gitVersion']（即集群版本），失败则回退到kubernetes_validate.latest_version()；
• strict=True时，意外的多余属性会被视为错误；
• 返回(warnings, errors)二元组：warnings中包括"找不到对应 schema（可能为自定义资源）"的提示，errors中包括校验错误与"Kubernetes 版本不受支持"等。

十、实战：完整示例串讲

10.1 ConfigMap 增删改查

完整代码见 configmap.py，流程为：创建（create）→ 按名字 + label 选择器列出（get）→ 修改 data 后patch→delete。其get返回的对象configmap_list.metadata.name、configmap_list.data展示了ResourceInstance的点号访问。

10.2 Deployment 滚动重启

完整代码见 deployment_rolling_restart.py：创建 3 副本 nginx Deployment 后，通过patch修改spec.template.metadata.annotations，写入kubectl.kubernetes.io/restartedAt时间戳触发滚动重启，最后删除。这是动态客户端驱动控制器式运维的典型例子。

10.3 Node 集群级查询

完整代码见 node.py：client.resources.get(api_version="v1", kind="Node")后列出所有节点，并对每个节点二次get，读取node.status.nodeInfo.kubeProxyVersion等字段。

10.4 Namespaced / Cluster CRD 全流程

完整代码见 namespaced_custom_resource.py 与 cluster_scoped_custom_resource.py。两者演示了动态客户端最核心的价值：

1. 通过apiextensions.k8s.io/v1的CustomResourceDefinition资源创建 CRD；
2. 捕获ResourceNotFoundError并等待约 2 秒，等待 discovery 更新后重新resources.get获取自定义资源 API；
3. 对自定义资源执行create/get（列表）/patch（指定application/merge-patch+json）/delete；
4. 最后删除 CRD。

两者的差别在于 CRDspec.scope为Namespaced（需要 namespace）还是Cluster（不需要 namespace）。

10.5 自定义 Accept 头与请求超时

accept_header.py 通过header_params传递Accept: application/json;as=PartialObjectMetadataList;v=v1;g=meta.k8s.io，演示如何请求部分对象元数据；request_timeout.py 演示每次调用传入_request_time=60设置客户端超时。

十一、测试与可靠性佐证

模块自带 e2e 测试 client_test.py，覆盖：

• 集群级与命名空间级自定义资源的完整生命周期（含 watch 超时行为、缓存失效后资源不可见）；
• Service、ReplicationController、ConfigMap、Node 等内置资源的 CRUD 与propagation_policy、pretty、label_selector等参数；
• PartialObjectMetadata（自定义 Accept 头）与 Server-Side Apply（field_manager、dry_run）；
• serialize_body对 dict / ResourceInstance / ResourceField 的序列化一致性。

这些测试直接佐证了本文描述的调用签名、参数语义与异常行为，是动手前值得通读的"活文档"。

十二、使用注意事项小结

1. 必须异步初始化：DynamicClient(apic)需要await（或async with），否则 discoverer 未初始化，访问resources/version会失败。
2. CRD 刚创建后需等待 discovery 刷新：捕获ResourceNotFoundError后重试，或调用client.resources.invalidate_cache()强制刷新。
3. namespaced 资源必须提供 namespace：可显式传namespace=，或在body['metadata']['namespace']中声明，否则抛ValueError。
4. patch CRD 要显式指定 content_type：CRD 无策略合并语义，使用application/merge-patch+json或application/json-patch+json；内置资源才适用默认的strategic-merge-patch+json。
5. delete 的选择器要求：至少提供name/label_selector/field_selector之一；namespaced 资源还需 namespace 或选择器。
6. watch 记得设置 timeout 或主动 stop：timeout=None时流不自动结束。
7. validate 为可选能力：依赖kubernetes_validate包，未安装时相关调用会抛KubernetesValidateMissing。

相关资源索引

• 模块源码：kubernetes/aio/dynamic/client.py、resource.py、discovery.py、exceptions.py
• 官方示例：examples_asyncio/dynamic-client/
• e2e 测试：kubernetes/aio/dynamic/client_test.py
• 同步版动态客户端参考：kubernetes/dynamic/

• 后端
• 云原生
• 容器编排