ConnectorTypeK8S

class connector_types.connector_type_k8s.ConnectorTypeK8S

Interact with a kubernetes cluster.

This connector type enables you to interact with a remote kubernetes cluster. See the Official Python client library for kubernetes for a description of available APIs and methods.

Inputs

Name	Type	Default	Description
api	str	None	The API to use. E.g. "CoreV1Api". Either document or (api and method) must be specified
args	list	None	Positional arguments to pass to the method call
certificate	str		The CA certificate of the kubernetes cluster
document	dict	None	A Kubernetes API object to apply. Either document or (api and method) must be specified
grant	str	None	the name of an oauth grant to use for authentication. Either token, key, or grant must be specified
host	str		The hostname or IP of the kubernetes cluster to use
key	dict	None	The content of the key file of the service account to use. Either token, key, or grant must be specified
kwargs	dict	None	Keyword arguments to pass to the method call
method	str	None	The method to call. E.g. "create_namespaced_deployment". Either document or (api and method) must be specified
parameters	dict	None	Deprecated: Additional keyword parameters to pass to the method call
replace	bool	False	If set to true, the resource will be replaced. Otherwise it will be patched, which means artifacts that have not changed will remain
token	str	None	The bearer token to use for authentication. Either token, key, or grant must be specified

Outputs

Name	Type	Default	Description
execution_id	int		The ID of the connection execution
message	str		The ended message for the connection. If the connection ended with an error, the message will contain information about what went wrong
result	dict
status	str		The ended status for the connection. Either "success" or "error".

Constants

input_list = ['api', 'args', 'certificate', 'document', 'grant', 'host', 'key', 'kwargs', 'method', 'parameters', 'replace', 'token'] output_list = ['result'] ssl_context_inputs = ['check_hostname', 'client_cert', 'client_key', 'server_ca'] version = 1

Methods

execute ()

log (message)

one_of_inputs (options)

run ()

Example

import yaml
import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    nginx_deployment_str = system.file('nginx-deployment.yaml').get_text_content()
    nginx_deployment = yaml.safe_load(nginx_deployment_str)
    this.connect(
        connector_type='K8S',
        token='*secret*',
        host='my-kubernetes-cluster-master',
        certificate='-----BEGIN CERTIFICATE-----\nMII...',
        api='ExtensionsV1beta1Api',
        method='create_namespaced_deployment',
        kwargs={
            'body': nginx_deployment,
        },
    )

    return this.success('all done')