目前还没有现成的API。

import json
from kubernetes import client, config

class ThingyApi(object):
    def __init__(self):
        config = client.Configuration()
        if not config.api_client:
            config.api_client = client.ApiClient()
        self.api_client = config.api_client

    def create_thingy(self, body, namespace='default'):
        resource_path = '/apis/example.com/v1/namespaces/' + namespace + '/thingies'
        header_params = {}
        header_params['Accept'] = self.api_client.select_header_accept(['application/json'])
        header_params['Content-Type'] = self.api_client.select_header_content_type(['*/*'])

        (resp, code, header) = self.api_client.call_api(
                resource_path, 'POST', {'namespace': namespace}, {}, header_params, body, [], _preload_content=False)

        return json.loads(resp.data.decode('utf-8'))

config.load_kube_config()

# Create the ThirdPartyResource (Thingy.example.com)
resp = client.ExtensionsV1beta1Api().create_third_party_resource(body={
    'apiVersion': 'extensions/v1beta1',
    'kind': 'ThirdPartyResource',
    'metadata': {'name': 'thingy.example.com'},
    'description': 'For storage of Thingy objects',
    'versions': [{'name': 'v1'}]})
print("ThirdPartyResource created")
print(str(resp))

# Create 3 Thingy objects (mything-{1,2,3})
thingyapi = ThingyApi()
for i in range(1, 4):
    obj = {'apiVersion': 'example.com/v1',
           'metadata': {'name': 'mythingy-'+str(i)},
           'kind': 'Thingy',
           # Arbitrary contents
           'key': 'value',
           'array': [40, 2],
           'object': {'foo': 'bar'}}
    resp = thingyapi.create_thingy(body=obj, namespace='default')
    print(str(resp))

字符串
输出将是这样的：

$ bin/python test.py                                                                                                                                                                    
ThirdPartyResource created
{'api_version': 'extensions/v1beta1',
 'description': 'For storage of Thingy objects',
 'kind': 'ThirdPartyResource',
 'metadata': {'annotations': None,
              'cluster_name': None,
              'creation_timestamp': u'2017-03-14T13:57:07Z',
              'deletion_grace_period_seconds': None,
              'deletion_timestamp': None,
              'finalizers': None,
              'generate_name': None,
              'generation': None,
              'labels': None,
              'name': 'thingy.example.com',
              'namespace': None,
              'owner_references': None,
              'resource_version': '59942',
              'self_link': '/apis/extensions/v1beta1/thirdpartyresourcesthingy.example.com',
              'uid': '1c596824-08be-11e7-9a5f-5254000f561a'},
 'versions': [{'name': 'v1'}]}
{u'kind': u'Thingy', u'object': {u'foo': u'bar'}, u'apiVersion': u'example.com/v1', u'key': u'value', u'array': [40, 2], u'metadata': {u'name': u'mythingy-1', u'namespace': u'default', u'resourceVersion': u'59943', u'creationTimestamp': u'2017-03-14T13:57:07Z', u'selfLink': u'/apis/example.com/v1/namespaces/default/thingies/mythingy-1', u'uid': u'1c59f7ae-08be-11e7-9a5f-5254000f561a'}}
{u'kind': u'Thingy', u'object': {u'foo': u'bar'}, u'apiVersion': u'example.com/v1', u'key': u'value', u'array': [40, 2], u'metadata': {u'name': u'mythingy-2', u'namespace': u'default', u'resourceVersion': u'59944', u'creationTimestamp': u'2017-03-14T13:57:07Z', u'selfLink': u'/apis/example.com/v1/namespaces/default/thingies/mythingy-2', u'uid': u'1c5be2a7-08be-11e7-9a5f-5254000f561a'}}
{u'kind': u'Thingy', u'object': {u'foo': u'bar'}, u'apiVersion': u'example.com/v1', u'key': u'value', u'array': [40, 2], u'metadata': {u'name': u'mythingy-3', u'namespace': u'default', u'resourceVersion': u'59945', u'creationTimestamp': u'2017-03-14T13:57:07Z', u'selfLink': u'/apis/example.com/v1/namespaces/default/thingies/mythingy-3', u'uid': u'1c5c390e-08be-11e7-9a5f-5254000f561a'}}

型
别忘了运行这个：

kubectl delete thingy --all
kubectl delete thirdpartyresource thingy.example.com

型

client-python现在支持TPR。这是the example from the repo：

from __future__ import print_function

from pprint import pprint

import kubernetes
from kubernetes import config
from kubernetes.rest import ApiException

config.load_kube_config()
api_instance = kubernetes.ThirdPartyResources()

namespace = 'default'
resource = 'repos'
fqdn = 'git.k8s.com'

body = {}
body['apiVersion'] = "git.k8s.com/v1"
body['kind'] = "RePo"
body['metadata'] = {}
body['metadata']['name'] = "blog-repo"
body['repo'] = "github.com/user/my-blog"
body['username'] = "username"
body['password'] = "password"
body['branch'] = "branch"


try: 
    # Create a Resource
    api_response = api_instance.apis_fqdn_v1_namespaces_namespace_resource_post(
        namespace, fqdn, resource, body)
    pprint(api_response)
except ApiException as e:
    print(
        "Exception when calling DefaultApi->apis_fqdn_v1_namespaces_namespace_resource_post: %s\n" % 
        e)

字符串
Ref：https://github.com/kubernetes-incubator/client-python/blob/master/examples/create_thirdparty_resource.md

**Update2024：**使用python SDK在kubernetes中创建自定义对象或CRD有两种方法：
***普通客户端：**根据Kubernetes API Swagger规范生成。它提供了一个强类型的API，并提供了特定于每个资源类型的方法，可以轻松地与API进行类型安全的交互。但它需要为每个资源类型预定义客户端对象。

namespace custom resource creation的代码片段：

from kubernetes import client, config
# ...(your function definition or your main function)
config.load_kube_config()

api = client.CustomObjectsApi()

# it's my custom resource defined as Dict
my_resource = {
    "apiVersion": "stable.example.com/v1",
    "kind": "CronTab",
    "metadata": {"name": "my-new-cron-object"},
    "spec": {
        "cronSpec": "* * * * */5",
        "image": "my-awesome-cron-image"
    }
}

# patch to update the `spec.cronSpec` field
patch_body = {
    "spec": {"cronSpec": "* * * * */10", "image": "my-awesome-cron-image"}
}

# create the resource
api.create_namespaced_custom_object(
    group="stable.example.com",
    version="v1",
    namespace="default",
    plural="crontabs",
    body=my_resource,
)
print("Resource created")

字符串

***动态客户端：**提供了一种更灵活、更动态的方式来与Kubernetes API进行交互。它提供了一个较低级别的接口，允许您使用动态对象和非结构化数据与API进行交互。它可能适合您创建CRD的用例。

代码片段：namespaced_custom_resource.py

from kubernetes import config, dynamic
from kubernetes.client import api_client
# ...(your function definition or your main function)
# Creating a dynamic client
client = dynamic.DynamicClient(
    api_client.ApiClient(configuration=config.load_kube_config())
)
# get ingressroute api
ingressroute_api = client.resources.get(
    api_version="apps.example.com/v1", kind="IngressRoute"
)
# define resource to be created
ingressroute_manifest = {
    "apiVersion": "apps.example.com/v1",
    "kind": "IngressRoute",
    "metadata": {
        "name": "ingress-route-first",
        "namespace": mynamespace,
    },
    "spec": {
        "virtualhost": {
            "fqdn": "www.google.com",
            "tls": {"secretName": "google-tls"},
        },
        "strategy": "RoundRobin",
    },
}
# create resource
ingressroute_api.create(body=ingressroute_manifest, namespace=mynamespace)

型
参考文献：

• https://github.com/kubernetes/client-go/issues/585的
• https://medium.com/cloud-native-daily/working-with-kubernetes-using-golang-a3069d51dfd6的