Cannot import StatefulSet

[TANguyen1893]

I am getting this import error from the logs:

Traceback (most recent call last): File "/usr/local/bin/deployment-orchestrator", line 6, in <module> from k8s_deployment_tools.deployment_orchestrator import main File "/usr/local/lib/python3.4/dist-packages/k8s_deployment_tools/deployment_orchestrator.py", line 31, in <module> from k8s_deployment_tools.deploy.operations.operation_detector import operation_detector File "/usr/local/lib/python3.4/dist-packages/k8s_deployment_tools/deploy/operations/operation_detector.py", line 20, in <module> from k8s_deployment_tools.deploy.operations.creator import creator File "/usr/local/lib/python3.4/dist-packages/k8s_deployment_tools/deploy/operations/creator.py", line 10, in <module> from pykube.objects import StatefulSet ImportError: cannot import name 'StatefulSet'

And my code is simply this:

from k8s_deployment_tools.deploy.operations.operation import operation
from pykube.http import HTTPClient
from pykube.config import KubeConfig
from pykube.objects import StatefulSet

import logging
from k8s_deployment_tools.utils.manifest import get_kind

class creator(operation):

    def __init__(self, manifest, rest_api_base_url):
        logging.info("REST API BASE URL")
        logging.info(rest_api_base_url)
        self.api_url = rest_api_base_url
        self.manifest_file = manifest
        self.api = HTTPClient(
            KubeConfig.from_url(rest_api_base_url))

    def perform(self):
        logging.info("CALLING PERFORM FROM CREATOR")
        kind = get_kind(self.manifest_file)
        pykube_object = self.get_pykube_object(kind, self.manifest_file)
        pykube_object.create()
        #TO-DO: poll until all pods are created

    def get_pykube_object(self, key, manifest_dict):
        k8s_objects = {
            # 'configmap': ConfigMap(self.api, manifest_dict),
            # 'daemonset': DaemonSet(self.api, manifest_dict),
            # 'deployment': Deployment(self.api, manifest_dict),
            # 'endpoint': Endpoint(self.api, manifest_dict),
            # 'event': Event(self.api, manifest_dict),
            # 'resourcequota': ResourceQuota(self.api, manifest_dict),
            # 'serviceaccount': ServiceAccount(self.api, manifest_dict),
            # 'ingress': Ingress(self.api, manifest_dict),
            # 'thirdpartyresource': ThirdPartyResource(self.api, manifest_dict),
            # 'job': Job(self.api, manifest_dict),
            # 'namespace': Namespace(self.api, manifest_dict),
            # 'node': Node(self.api, manifest_dict),
            # 'pod': Pod(self.api, manifest_dict),
            # 'replicationcontroller': ReplicationController(self.api, manifest_dict),
            # 'replicaset': ReplicaSet(self.api, manifest_dict),
            # 'secret': Secret(self.api, manifest_dict),
            # 'service': Service(self.api, manifest_dict),
            # 'persistentvolume': PersistentVolume(self.api, manifest_dict),
            # 'persistentvolumeclaim': PersistentVolumeClaim(self.api, manifest_dict),
            # 'horizontalpodautoscaler': HorizontalPodAutoscaler(self.api, manifest_dict),
            # 'petset': PetSet(self.api, manifest_dict),
            # 'statefulset': StatefulSet(self.api, manifest_dict),
            # 'role': Role(self.api, manifest_dict),
            # 'rolebinding': RoleBinding(self.api, manifest_dict),
            # 'clusterrole': ClusterRole(self.api, manifest_dict),
            # 'clusterrolebinding': ClusterRoleBinding(self.api, manifest_dict),
            # 'podsecuritypolicy': PodSecurityPolicy(self.api, manifest_dict)

        }
        return k8s_objects[key]

[TANguyen1893]

My requirements.txt had the wrong version of pykube. My mistake, closing this issue.