Module pachyderm_sdk.api.license

Classes

class ActivateRequest (activation_code: str = None, expires: datetime.datetime = None)

ActivateRequest(activation_code: str = None, expires: datetime.datetime = None)

Expand source code

@dataclass(eq=False, repr=False)
class ActivateRequest(betterproto.Message):
    activation_code: str = betterproto.string_field(1)
    """
    activation_code is a Pachyderm enterprise activation code. New users can
    obtain trial activation codes
    """

    expires: datetime = betterproto.message_field(2)
    """
    expires is a timestamp indicating when this activation code will expire.
    This should not generally be set (it's primarily used for testing), and is
    only applied if it's earlier than the signed expiration time in
    'activation_code'.
    """

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var activation_code : str

activation_code is a Pachyderm enterprise activation code. New users can obtain trial activation codes

var expires : datetime.datetime

expires is a timestamp indicating when this activation code will expire. This should not generally be set (it's primarily used for testing), and is only applied if it's earlier than the signed expiration time in 'activation_code'.

class ActivateResponse (info: _enterprise__.TokenInfo = None)

ActivateResponse(info: '_enterprise__.TokenInfo' = None)

Expand source code

@dataclass(eq=False, repr=False)
class ActivateResponse(betterproto.Message):
    info: "_enterprise__.TokenInfo" = betterproto.message_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var info : TokenInfo

class GetActivationCodeRequest

GetActivationCodeRequest()

Expand source code

@dataclass(eq=False, repr=False)
class GetActivationCodeRequest(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class GetActivationCodeResponse (state: _enterprise__.State = None, info: _enterprise__.TokenInfo = None, activation_code: str = None)

GetActivationCodeResponse(state: '_enterprise__.State' = None, info: '_enterprise__.TokenInfo' = None, activation_code: str = None)

Expand source code

@dataclass(eq=False, repr=False)
class GetActivationCodeResponse(betterproto.Message):
    state: "_enterprise__.State" = betterproto.enum_field(1)
    info: "_enterprise__.TokenInfo" = betterproto.message_field(2)
    activation_code: str = betterproto.string_field(3)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var state : State

var info : TokenInfo

var activation_code : str

class DeactivateRequest

DeactivateRequest()

Expand source code

@dataclass(eq=False, repr=False)
class DeactivateRequest(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class DeactivateResponse

DeactivateResponse()

Expand source code

@dataclass(eq=False, repr=False)
class DeactivateResponse(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class AddClusterRequest (id: str = None, address: str = None, secret: str = None, user_address: str = None, cluster_deployment_id: str = None, enterprise_server: bool = None)

AddClusterRequest(id: str = None, address: str = None, secret: str = None, user_address: str = None, cluster_deployment_id: str = None, enterprise_server: bool = None)

Expand source code

@dataclass(eq=False, repr=False)
class AddClusterRequest(betterproto.Message):
    id: str = betterproto.string_field(1)
    """id is the unique, immutable identifier for this cluster"""

    address: str = betterproto.string_field(2)
    """address is the public GPRC address where the cluster can be reached"""

    secret: str = betterproto.string_field(3)
    """
    If set, secret specifies the shared secret this cluster will use to
    authenticate to the license server. Otherwise a secret will be generated
    and returned in the response.
    """

    user_address: str = betterproto.string_field(4)
    """The pachd address for users to connect to."""

    cluster_deployment_id: str = betterproto.string_field(5)
    """The deployment ID value generated by each Cluster"""

    enterprise_server: bool = betterproto.bool_field(6)
    """
    This field indicates whether the address points to an enterprise server
    """

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

id is the unique, immutable identifier for this cluster

var address : str

address is the public GPRC address where the cluster can be reached

var secret : str

If set, secret specifies the shared secret this cluster will use to authenticate to the license server. Otherwise a secret will be generated and returned in the response.

var user_address : str

The pachd address for users to connect to.

var cluster_deployment_id : str

The deployment ID value generated by each Cluster

var enterprise_server : bool

This field indicates whether the address points to an enterprise server

class AddClusterResponse (secret: str = None)

AddClusterResponse(secret: str = None)

Expand source code

@dataclass(eq=False, repr=False)
class AddClusterResponse(betterproto.Message):
    secret: str = betterproto.string_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var secret : str

class DeleteClusterRequest (id: str = None)

DeleteClusterRequest(id: str = None)

Expand source code

@dataclass(eq=False, repr=False)
class DeleteClusterRequest(betterproto.Message):
    id: str = betterproto.string_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

class DeleteClusterResponse

DeleteClusterResponse()

Expand source code

@dataclass(eq=False, repr=False)
class DeleteClusterResponse(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class ClusterStatus (id: str = None, address: str = None, version: str = None, auth_enabled: bool = None, client_id: str = None, last_heartbeat: datetime.datetime = None, created_at: datetime.datetime = None)

ClusterStatus(id: str = None, address: str = None, version: str = None, auth_enabled: bool = None, client_id: str = None, last_heartbeat: datetime.datetime = None, created_at: datetime.datetime = None)

Expand source code

@dataclass(eq=False, repr=False)
class ClusterStatus(betterproto.Message):
    id: str = betterproto.string_field(1)
    address: str = betterproto.string_field(2)
    version: str = betterproto.string_field(3)
    auth_enabled: bool = betterproto.bool_field(4)
    client_id: str = betterproto.string_field(7)
    last_heartbeat: datetime = betterproto.message_field(5)
    created_at: datetime = betterproto.message_field(6)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

var address : str

var version : str

var auth_enabled : bool

var client_id : str

var last_heartbeat : datetime.datetime

var created_at : datetime.datetime

class UpdateClusterRequest (id: str = None, address: str = None, user_address: str = None, cluster_deployment_id: str = None, secret: str = None)

Note: Updates of the enterprise-server field are not allowed. In the worst case, a user can recreate their cluster if they need the field updated.

Expand source code

@dataclass(eq=False, repr=False)
class UpdateClusterRequest(betterproto.Message):
    """
    Note: Updates of the enterprise-server field are not allowed. In the worst
    case, a user can recreate their cluster if they need the field updated.
    """

    id: str = betterproto.string_field(1)
    address: str = betterproto.string_field(2)
    user_address: str = betterproto.string_field(3)
    cluster_deployment_id: str = betterproto.string_field(4)
    secret: str = betterproto.string_field(5)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

var address : str

var user_address : str

var cluster_deployment_id : str

var secret : str

class UpdateClusterResponse

UpdateClusterResponse()

Expand source code

@dataclass(eq=False, repr=False)
class UpdateClusterResponse(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class ListClustersRequest

ListClustersRequest()

Expand source code

@dataclass(eq=False, repr=False)
class ListClustersRequest(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class ListClustersResponse (clusters: List[ForwardRef('ClusterStatus')] = None)

ListClustersResponse(clusters: List[ForwardRef('ClusterStatus')] = None)

Expand source code

@dataclass(eq=False, repr=False)
class ListClustersResponse(betterproto.Message):
    clusters: List["ClusterStatus"] = betterproto.message_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var clusters : List[ClusterStatus]

class DeleteAllRequest

DeleteAllRequest()

Expand source code

@dataclass(eq=False, repr=False)
class DeleteAllRequest(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class DeleteAllResponse

DeleteAllResponse()

Expand source code

@dataclass(eq=False, repr=False)
class DeleteAllResponse(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class HeartbeatRequest (id: str = None, secret: str = None, version: str = None, auth_enabled: bool = None, client_id: str = None)

HeartbeatRequest(id: str = None, secret: str = None, version: str = None, auth_enabled: bool = None, client_id: str = None)

Expand source code

@dataclass(eq=False, repr=False)
class HeartbeatRequest(betterproto.Message):
    id: str = betterproto.string_field(1)
    secret: str = betterproto.string_field(2)
    version: str = betterproto.string_field(3)
    auth_enabled: bool = betterproto.bool_field(4)
    client_id: str = betterproto.string_field(5)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

var secret : str

var version : str

var auth_enabled : bool

var client_id : str

class HeartbeatResponse (license: _enterprise__.LicenseRecord = None)

HeartbeatResponse(license: '_enterprise__.LicenseRecord' = None)

Expand source code

@dataclass(eq=False, repr=False)
class HeartbeatResponse(betterproto.Message):
    license: "_enterprise__.LicenseRecord" = betterproto.message_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var license : LicenseRecord

class UserClusterInfo (id: str = None, cluster_deployment_id: str = None, address: str = None, enterprise_server: bool = None)

UserClusterInfo(id: str = None, cluster_deployment_id: str = None, address: str = None, enterprise_server: bool = None)

Expand source code

@dataclass(eq=False, repr=False)
class UserClusterInfo(betterproto.Message):
    id: str = betterproto.string_field(1)
    cluster_deployment_id: str = betterproto.string_field(2)
    address: str = betterproto.string_field(3)
    enterprise_server: bool = betterproto.bool_field(4)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var id : str

var cluster_deployment_id : str

var address : str

var enterprise_server : bool

class ListUserClustersRequest

ListUserClustersRequest()

Expand source code

@dataclass(eq=False, repr=False)
class ListUserClustersRequest(betterproto.Message):
    pass

Ancestors

• betterproto.Message
• abc.ABC

class ListUserClustersResponse (clusters: List[ForwardRef('UserClusterInfo')] = None)

ListUserClustersResponse(clusters: List[ForwardRef('UserClusterInfo')] = None)

Expand source code

@dataclass(eq=False, repr=False)
class ListUserClustersResponse(betterproto.Message):
    clusters: List["UserClusterInfo"] = betterproto.message_field(1)

Ancestors

• betterproto.Message
• abc.ABC

Class variables

var clusters : List[UserClusterInfo]

class ApiStub (channel: grpc.Channel)

Expand source code

class ApiStub:
    def __init__(self, channel: "grpc.Channel"):
        self.__rpc_activate = channel.unary_unary(
            "/license_v2.API/Activate",
            request_serializer=ActivateRequest.SerializeToString,
            response_deserializer=ActivateResponse.FromString,
        )
        self.__rpc_get_activation_code = channel.unary_unary(
            "/license_v2.API/GetActivationCode",
            request_serializer=GetActivationCodeRequest.SerializeToString,
            response_deserializer=GetActivationCodeResponse.FromString,
        )
        self.__rpc_delete_all = channel.unary_unary(
            "/license_v2.API/DeleteAll",
            request_serializer=DeleteAllRequest.SerializeToString,
            response_deserializer=DeleteAllResponse.FromString,
        )
        self.__rpc_add_cluster = channel.unary_unary(
            "/license_v2.API/AddCluster",
            request_serializer=AddClusterRequest.SerializeToString,
            response_deserializer=AddClusterResponse.FromString,
        )
        self.__rpc_delete_cluster = channel.unary_unary(
            "/license_v2.API/DeleteCluster",
            request_serializer=DeleteClusterRequest.SerializeToString,
            response_deserializer=DeleteClusterResponse.FromString,
        )
        self.__rpc_list_clusters = channel.unary_unary(
            "/license_v2.API/ListClusters",
            request_serializer=ListClustersRequest.SerializeToString,
            response_deserializer=ListClustersResponse.FromString,
        )
        self.__rpc_update_cluster = channel.unary_unary(
            "/license_v2.API/UpdateCluster",
            request_serializer=UpdateClusterRequest.SerializeToString,
            response_deserializer=UpdateClusterResponse.FromString,
        )
        self.__rpc_heartbeat = channel.unary_unary(
            "/license_v2.API/Heartbeat",
            request_serializer=HeartbeatRequest.SerializeToString,
            response_deserializer=HeartbeatResponse.FromString,
        )
        self.__rpc_list_user_clusters = channel.unary_unary(
            "/license_v2.API/ListUserClusters",
            request_serializer=ListUserClustersRequest.SerializeToString,
            response_deserializer=ListUserClustersResponse.FromString,
        )

    def activate(
        self, *, activation_code: str = "", expires: datetime = None
    ) -> "ActivateResponse":
        request = ActivateRequest()
        request.activation_code = activation_code
        if expires is not None:
            request.expires = expires

        return self.__rpc_activate(request)

    def get_activation_code(self) -> "GetActivationCodeResponse":
        request = GetActivationCodeRequest()

        return self.__rpc_get_activation_code(request)

    def delete_all(self) -> "DeleteAllResponse":
        request = DeleteAllRequest()

        return self.__rpc_delete_all(request)

    def add_cluster(
        self,
        *,
        id: str = "",
        address: str = "",
        secret: str = "",
        user_address: str = "",
        cluster_deployment_id: str = "",
        enterprise_server: bool = False
    ) -> "AddClusterResponse":
        request = AddClusterRequest()
        request.id = id
        request.address = address
        request.secret = secret
        request.user_address = user_address
        request.cluster_deployment_id = cluster_deployment_id
        request.enterprise_server = enterprise_server

        return self.__rpc_add_cluster(request)

    def delete_cluster(self, *, id: str = "") -> "DeleteClusterResponse":
        request = DeleteClusterRequest()
        request.id = id

        return self.__rpc_delete_cluster(request)

    def list_clusters(self) -> "ListClustersResponse":
        request = ListClustersRequest()

        return self.__rpc_list_clusters(request)

    def update_cluster(
        self,
        *,
        id: str = "",
        address: str = "",
        user_address: str = "",
        cluster_deployment_id: str = "",
        secret: str = ""
    ) -> "UpdateClusterResponse":
        request = UpdateClusterRequest()
        request.id = id
        request.address = address
        request.user_address = user_address
        request.cluster_deployment_id = cluster_deployment_id
        request.secret = secret

        return self.__rpc_update_cluster(request)

    def heartbeat(
        self,
        *,
        id: str = "",
        secret: str = "",
        version: str = "",
        auth_enabled: bool = False,
        client_id: str = ""
    ) -> "HeartbeatResponse":
        request = HeartbeatRequest()
        request.id = id
        request.secret = secret
        request.version = version
        request.auth_enabled = auth_enabled
        request.client_id = client_id

        return self.__rpc_heartbeat(request)

    def list_user_clusters(self) -> "ListUserClustersResponse":
        request = ListUserClustersRequest()

        return self.__rpc_list_user_clusters(request)

Methods

def activate(self, *, activation_code: str = '', expires: datetime.datetime = None) ‑> ActivateResponse

Expand source code

def activate(
    self, *, activation_code: str = "", expires: datetime = None
) -> "ActivateResponse":
    request = ActivateRequest()
    request.activation_code = activation_code
    if expires is not None:
        request.expires = expires

    return self.__rpc_activate(request)

def get_activation_code(self) ‑> GetActivationCodeResponse

Expand source code

def get_activation_code(self) -> "GetActivationCodeResponse":
    request = GetActivationCodeRequest()

    return self.__rpc_get_activation_code(request)

def delete_all(self) ‑> DeleteAllResponse

Expand source code

def delete_all(self) -> "DeleteAllResponse":
    request = DeleteAllRequest()

    return self.__rpc_delete_all(request)

def add_cluster(self, *, id: str = '', address: str = '', secret: str = '', user_address: str = '', cluster_deployment_id: str = '', enterprise_server: bool = False) ‑> AddClusterResponse

Expand source code

def add_cluster(
    self,
    *,
    id: str = "",
    address: str = "",
    secret: str = "",
    user_address: str = "",
    cluster_deployment_id: str = "",
    enterprise_server: bool = False
) -> "AddClusterResponse":
    request = AddClusterRequest()
    request.id = id
    request.address = address
    request.secret = secret
    request.user_address = user_address
    request.cluster_deployment_id = cluster_deployment_id
    request.enterprise_server = enterprise_server

    return self.__rpc_add_cluster(request)

def delete_cluster(self, *, id: str = '') ‑> DeleteClusterResponse

Expand source code

def delete_cluster(self, *, id: str = "") -> "DeleteClusterResponse":
    request = DeleteClusterRequest()
    request.id = id

    return self.__rpc_delete_cluster(request)

def list_clusters(self) ‑> ListClustersResponse

Expand source code

def list_clusters(self) -> "ListClustersResponse":
    request = ListClustersRequest()

    return self.__rpc_list_clusters(request)

def update_cluster(self, *, id: str = '', address: str = '', user_address: str = '', cluster_deployment_id: str = '', secret: str = '') ‑> UpdateClusterResponse

Expand source code

def update_cluster(
    self,
    *,
    id: str = "",
    address: str = "",
    user_address: str = "",
    cluster_deployment_id: str = "",
    secret: str = ""
) -> "UpdateClusterResponse":
    request = UpdateClusterRequest()
    request.id = id
    request.address = address
    request.user_address = user_address
    request.cluster_deployment_id = cluster_deployment_id
    request.secret = secret

    return self.__rpc_update_cluster(request)

def heartbeat(self, *, id: str = '', secret: str = '', version: str = '', auth_enabled: bool = False, client_id: str = '') ‑> HeartbeatResponse

Expand source code

def heartbeat(
    self,
    *,
    id: str = "",
    secret: str = "",
    version: str = "",
    auth_enabled: bool = False,
    client_id: str = ""
) -> "HeartbeatResponse":
    request = HeartbeatRequest()
    request.id = id
    request.secret = secret
    request.version = version
    request.auth_enabled = auth_enabled
    request.client_id = client_id

    return self.__rpc_heartbeat(request)

def list_user_clusters(self) ‑> ListUserClustersResponse

Expand source code

def list_user_clusters(self) -> "ListUserClustersResponse":
    request = ListUserClustersRequest()

    return self.__rpc_list_user_clusters(request)

class ApiBase

Expand source code

class ApiBase:
    def activate(
        self, activation_code: str, expires: datetime, context: "grpc.ServicerContext"
    ) -> "ActivateResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def get_activation_code(
        self, context: "grpc.ServicerContext"
    ) -> "GetActivationCodeResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_all(self, context: "grpc.ServicerContext") -> "DeleteAllResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def add_cluster(
        self,
        id: str,
        address: str,
        secret: str,
        user_address: str,
        cluster_deployment_id: str,
        enterprise_server: bool,
        context: "grpc.ServicerContext",
    ) -> "AddClusterResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def delete_cluster(
        self, id: str, context: "grpc.ServicerContext"
    ) -> "DeleteClusterResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def list_clusters(self, context: "grpc.ServicerContext") -> "ListClustersResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def update_cluster(
        self,
        id: str,
        address: str,
        user_address: str,
        cluster_deployment_id: str,
        secret: str,
        context: "grpc.ServicerContext",
    ) -> "UpdateClusterResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def heartbeat(
        self,
        id: str,
        secret: str,
        version: str,
        auth_enabled: bool,
        client_id: str,
        context: "grpc.ServicerContext",
    ) -> "HeartbeatResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    def list_user_clusters(
        self, context: "grpc.ServicerContext"
    ) -> "ListUserClustersResponse":
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")

    __proto_path__ = "license_v2.API"

    @property
    def __rpc_methods__(self):
        return {
            "Activate": grpc.unary_unary_rpc_method_handler(
                self.activate,
                request_deserializer=ActivateRequest.FromString,
                response_serializer=ActivateRequest.SerializeToString,
            ),
            "GetActivationCode": grpc.unary_unary_rpc_method_handler(
                self.get_activation_code,
                request_deserializer=GetActivationCodeRequest.FromString,
                response_serializer=GetActivationCodeRequest.SerializeToString,
            ),
            "DeleteAll": grpc.unary_unary_rpc_method_handler(
                self.delete_all,
                request_deserializer=DeleteAllRequest.FromString,
                response_serializer=DeleteAllRequest.SerializeToString,
            ),
            "AddCluster": grpc.unary_unary_rpc_method_handler(
                self.add_cluster,
                request_deserializer=AddClusterRequest.FromString,
                response_serializer=AddClusterRequest.SerializeToString,
            ),
            "DeleteCluster": grpc.unary_unary_rpc_method_handler(
                self.delete_cluster,
                request_deserializer=DeleteClusterRequest.FromString,
                response_serializer=DeleteClusterRequest.SerializeToString,
            ),
            "ListClusters": grpc.unary_unary_rpc_method_handler(
                self.list_clusters,
                request_deserializer=ListClustersRequest.FromString,
                response_serializer=ListClustersRequest.SerializeToString,
            ),
            "UpdateCluster": grpc.unary_unary_rpc_method_handler(
                self.update_cluster,
                request_deserializer=UpdateClusterRequest.FromString,
                response_serializer=UpdateClusterRequest.SerializeToString,
            ),
            "Heartbeat": grpc.unary_unary_rpc_method_handler(
                self.heartbeat,
                request_deserializer=HeartbeatRequest.FromString,
                response_serializer=HeartbeatRequest.SerializeToString,
            ),
            "ListUserClusters": grpc.unary_unary_rpc_method_handler(
                self.list_user_clusters,
                request_deserializer=ListUserClustersRequest.FromString,
                response_serializer=ListUserClustersRequest.SerializeToString,
            ),
        }

Methods

def activate(self, activation_code: str, expires: datetime.datetime, context: grpc.ServicerContext) ‑> ActivateResponse

Expand source code

def activate(
    self, activation_code: str, expires: datetime, context: "grpc.ServicerContext"
) -> "ActivateResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def get_activation_code(self, context: grpc.ServicerContext) ‑> GetActivationCodeResponse

Expand source code

def get_activation_code(
    self, context: "grpc.ServicerContext"
) -> "GetActivationCodeResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def delete_all(self, context: grpc.ServicerContext) ‑> DeleteAllResponse

Expand source code

def delete_all(self, context: "grpc.ServicerContext") -> "DeleteAllResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def add_cluster(self, id: str, address: str, secret: str, user_address: str, cluster_deployment_id: str, enterprise_server: bool, context: grpc.ServicerContext) ‑> AddClusterResponse

Expand source code

def add_cluster(
    self,
    id: str,
    address: str,
    secret: str,
    user_address: str,
    cluster_deployment_id: str,
    enterprise_server: bool,
    context: "grpc.ServicerContext",
) -> "AddClusterResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def delete_cluster(self, id: str, context: grpc.ServicerContext) ‑> DeleteClusterResponse

Expand source code

def delete_cluster(
    self, id: str, context: "grpc.ServicerContext"
) -> "DeleteClusterResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def list_clusters(self, context: grpc.ServicerContext) ‑> ListClustersResponse

Expand source code

def list_clusters(self, context: "grpc.ServicerContext") -> "ListClustersResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def update_cluster(self, id: str, address: str, user_address: str, cluster_deployment_id: str, secret: str, context: grpc.ServicerContext) ‑> UpdateClusterResponse

Expand source code

def update_cluster(
    self,
    id: str,
    address: str,
    user_address: str,
    cluster_deployment_id: str,
    secret: str,
    context: "grpc.ServicerContext",
) -> "UpdateClusterResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def heartbeat(self, id: str, secret: str, version: str, auth_enabled: bool, client_id: str, context: grpc.ServicerContext) ‑> HeartbeatResponse

Expand source code

def heartbeat(
    self,
    id: str,
    secret: str,
    version: str,
    auth_enabled: bool,
    client_id: str,
    context: "grpc.ServicerContext",
) -> "HeartbeatResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")

def list_user_clusters(self, context: grpc.ServicerContext) ‑> ListUserClustersResponse

Expand source code

def list_user_clusters(
    self, context: "grpc.ServicerContext"
) -> "ListUserClustersResponse":
    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
    context.set_details("Method not implemented!")
    raise NotImplementedError("Method not implemented!")