Interacting with Kubernetes Deployments and Services using Python SDK

delete_deployment.py

from kubernetes import client, config

def delete_deployment(api_instance, deployment_name, namespace):
    # Delete Deployment
    api_response = api_instance.delete_namespaced_deployment(
        name=deployment_name,
        namespace=namespace,
        body=client.V1DeleteOptions(propagation_policy="Foreground", grace_period_seconds=5)
    )
    print("Deployment deleted")

def main():
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()

    namespace = "default"
    deployment_name = "nginx-deployment"
    delete_deployment(apps_v1, deployment_name, namespace)

if __name__ == '__main__':
    main()

delete_service.py

from kubernetes import client, config

def delete_service(api_instance, service_name, namespace):
    # Delete Service
    api_response = api_instance.delete_namespaced_service(
        name=service_name,
        namespace=namespace,
        body=client.V1DeleteOptions()
    )
    print("Service deleted")

def main():
    config.load_kube_config()
    core_v1 = client.CoreV1Api()

    namespace = "default"
    service_name = "nginx-service"
    delete_service(core_v1, service_name, namespace)

if __name__ == '__main__':
    main()

list_deployment.py

from kubernetes import client, config

def list_deployments(api_instance, namespace):
    # List Deployments
    deployments = api_instance.list_namespaced_deployment(namespace)
    for deployment in deployments.items:
        print("Name: %s, Replicas: %s" % (deployment.metadata.name, deployment.spec.replicas))

def main():
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()
    namespace = "default"
    list_deployments(apps_v1, namespace)

if __name__ == '__main__':
    main()

list_service.py

from kubernetes import client, config

def list_services(api_instance, namespace):
    # List Services
    services = api_instance.list_namespaced_service(namespace)
    for service in services.items:
        print("Name: %s, Cluster IP: %s" % (service.metadata.name, service.spec.cluster_ip))

def main():
    config.load_kube_config()
    namespace = "default"
    core_v1 = client.CoreV1Api()
    list_services(core_v1, namespace)

if __name__ == '__main__':
    main()

update_deployment.py

from kubernetes import client, config

def update_deployment_replicas(api_instance, deployment_name, new_replicas, namespace):
    # Update Deployment replicas
    body = {
        "spec": {
            "replicas": new_replicas
        }
    }
    api_response = api_instance.patch_namespaced_deployment(
        name=deployment_name,
        namespace=namespace,
        body=body
    )
    print("Deployment updated")

def main():
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()

    namespace = "default"
    deployment_name = "nginx-deployment"
    new_replicas = 5
    update_deployment_replicas(apps_v1, deployment_name, new_replicas, namespace)

if __name__ == '__main__':
    main()

update_service.py

from kubernetes import client, config

def update_service_type(api_instance, service_name, new_type, namespace):
    # Update Service type
    body = {
        "spec": {
            "type": new_type
        }
    }
    api_response = api_instance.patch_namespaced_service(
        name=service_name,
        namespace=namespace,
        body=body
    )
    print("Service updated")

def main():
    config.load_kube_config()
    core_v1 = client.CoreV1Api()

    namespace = "default"
    service_name = "nginx-service"
    new_type = "NodePort"
    update_service_type(core_v1, service_name, new_type, namespace)

if __name__ == '__main__':
    main()