Skip to content

Client: Python

Here, we discuss some implementations of the python client of k8s, which may help you debug some errors

1. Configuration

1.1 Load Kube Config

Config

config.load_kube_config()

Here, it will try to load the config from a kubectl config file.

The file is default at path ~/.kube/config, you can also change the config file you want to use by:

  • pass the argument config.load_kube_config(config_file='path-to-config')
  • change the envrionment variable KUBECONFIG to path-to-config

But, how does it work underneath the code? Here is a flow chart of how the codes work:

stateDiagram-v2
    [*] --> load_kube_config
    config_file --> _get_kube_config_loader
    load_kube_config --> _get_kube_config_loader
    _get_kube_config_loader --> load_kube_config
    load_kube_config --> load_and_set
    load_and_set --> Configuration.set_default
    Configuration.set_default --> [*]

Now let's have a look at the source codes.

1. Load Kube Config

Source code is here

The first part is easy, it will call _get_kube_config_loader() to get a config loader, then call the loader's load_and_set(config) to load the config from file and set the attribute to the config object.

def load_kube_config(config_file=None, context=None,
                     client_configuration=None,
                     persist_config=True):
    if config_file is None:
        config_file = KUBE_CONFIG_DEFAULT_LOCATION

    loader = _get_kube_config_loader(
        filename=config_file, active_context=context,
        persist_config=persist_config)

    if client_configuration is None:
        config = type.__call__(Configuration)
        loader.load_and_set(config)
        Configuration.set_default(config)
    else:
        loader.load_and_set(client_configuration)

2. Load and Set

Source code is here

In load_and_set, the code will then load info and set the attributes of a client_configuration.

def load_and_set(self, client_configuration):
        self._load_authentication()
        self._load_cluster_info()
        self._set_config(client_configuration)

3. _set_config

Source code is here

In the _set_config, fields host, ssl_ca_cert, cert_file, key_file, verify_ssl will be set to loaded config value.

def _set_config(self, client_configuration):
    ...
    keys = ['host', 'ssl_ca_cert', 'cert_file', 'key_file', 'verify_ssl']
    for key in keys:
        if key in self.__dict__:
            setattr(client_configuration, key, getattr(self, key))

4. Set Default Config

Source code is here

After the configuration is set, in load_kube_config, Configuration.set_default(config) is called to save the current config to a class attribute _default

class Configuration(object):
    @classmethod
    def set_default(cls, default):
        cls._default = copy.deepcopy(default)

2. Client

2.1 Create an API Client

Create a Client

v1 = client.CoreV1Api()

How does it loads the correct config?

stateDiagram-v2
    [*] --> CoreV1Api()
    CoreV1Api() --> ApiClient()
    ApiClient() --> Configuration.get_default_copy()
    Configuration.get_default_copy() --> copy.deepcopy(cls._default)
    copy.deepcopy(cls._default) --> [*]

1. Create a Specific API Client Object

Source code is here

When a CoreV1Api is initializing, it will set self.api_client, and if not passed in, it will create an ApiClient instance.

class CoreV1Api(object):
    def __init__(self, api_client=None):
        if api_client is None:
            api_client = ApiClient()
        self.api_client = api_client

2. API Client Copy the Default Config

Source code is here

When ApiClient initializes, if no configuration is provided, it will call Configuration.get_default_copy() to get a configuration object.

class ApiClient(object):
    def __init__(self, configuration=None, header_name=None, header_value=None,
                cookie=None, pool_threads=1):
        if configuration is None:
            configuration = Configuration.get_default_copy()

Source code is here

In the Configuration.get_default_copy, it will return the deepcopy of the previously loaded config. And that is how it get the loaded configuration automatically.

@classmethod
def get_default_copy(cls):
    if cls._default is not None:
        return copy.deepcopy(cls._default)
    return Configuration()