Python Kubernetes Client: equivalent of kubectl api-resources --namespaced=false

Question

Via the CLI, I can use kubectl api-resources --namespaced=false to list all available cluster-scoped resources in a cluster.

I am writing a custom operator with the Python Kubernetes Client API, however I can't seem to find anything in the API that allows me to do this.

The closest I have found is the following code, which was included as an example in the repository:

from kubernetes import client, config


def main():
    # Configs can be set in Configuration class directly or using helper
    # utility. If no argument provided, the config will be loaded from
    # default location.
    config.load_kube_config()

    print("Supported APIs (* is preferred version):")
    print("%-40s %s" %
          ("core", ",".join(client.CoreApi().get_api_versions().versions)))
    for api in client.ApisApi().get_api_versions().groups:
        versions = []
        for v in api.versions:
            name = ""
            if v.version == api.preferred_version.version and len(
                    api.versions) > 1:
                name += "*"
            name += v.version
            versions.append(name)
        print("%-40s %s" % (api.name, ",".join(versions)))


if __name__ == '__main__':
    main()

Unfortunately, client.ApisApi() doesn't have a get_api_resources() option.

Does anybody know of a way that I can list all the api-resources?

Answer 1

I managed to solve the issue. For anyone wondering, the Python client API unfortunately does not have an equivalent function to kubectl api-resources, or kubectl api-resources --namespaced=false.

I was able to create my own solution to this using the API, which you can see below. To get similar output to kubectl api-resources --namespaced=false, you would just call getNonNamespacedApiResources. Note that this python program runs inside of a Pod associated with a ServiceAccount, which gives it the necessary permissions.

def query_api(api_group):
    SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token"
    SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
    KUBERNETES_HOST = "https://%s:%s" % (os.getenv("KUBERNETES_SERVICE_HOST"), os.getenv("KUBERNETES_SERVICE_PORT"))

    if not os.path.isfile(SERVICE_TOKEN_FILENAME):
        raise client.rest.ApiException("Service token file does not exists.")

    with open(SERVICE_TOKEN_FILENAME) as f:
        token = f.read()
        if not token:
            raise client.rest.ApiException("Token file exists but empty.")

        # set token bearer in header
        headers = {"Authorization": "Bearer " + token.strip('\n')}

    if not os.path.isfile(SERVICE_CERT_FILENAME):
        raise client.rest.ApiException("Service certification file does not exists.")

    with open(SERVICE_CERT_FILENAME) as f:
        if not f.read():
            raise client.rest.ApiException("Cert file exists but empty.")

    # query the API
    response = requests.get(
        KUBERNETES_HOST + "/apis/" + api_group, 
        headers=headers, 
        verify=config.incluster_config.SERVICE_CERT_FILENAME
    ).json()

    return response

def extractClusterScopedResources(api_group):
    api_response_json = query_api(api_group)

    scoped_resources = []

    for obj in api_response_json['resources']:
        if not obj['namespaced'] and "storageVersionHash" in obj:
            scoped_resources.append(obj['name'])

    return scoped_resources

def getNonNamespacedApiResources():
    apis = []

    # handle kubernetes/apis/
    for api in client.ApisApi().get_api_versions().groups:
        for v in api.versions:
            if v.version == api.preferred_version.version:
                apis.append(api.name + "/" + v.version)
                continue
    api_resources = []
    for api_group in apis:
        api_resources.extend(extractClusterScopedResources(api_group))

    # handle kubernetes/api/v1
    for api in client.CoreV1Api().get_api_resources().resources:
        if not api.namespaced and api.storage_version_hash:
            api_resources.append(api.name)

    return api_resources