# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: python_pachyderm/proto/v2/enterprise/enterprise.proto
# plugin: python-betterproto
from dataclasses import dataclass
from datetime import datetime
from typing import Dict
import betterproto
from betterproto.grpc.grpclib_server import ServiceBase
import grpclib
[docs]class State(betterproto.Enum):
NONE = 0
ACTIVE = 1
EXPIRED = 2
HEARTBEAT_FAILED = 3
[docs]class PauseStatusResponsePauseStatus(betterproto.Enum):
UNPAUSED = 0
PARTIALLY_PAUSED = 1
PAUSED = 2
[docs]@dataclass(eq=False, repr=False)
class LicenseRecord(betterproto.Message):
"""
LicenseRecord is the record we store in etcd for a Pachyderm enterprise
token that has been provided to a Pachyderm license server
"""
activation_code: str = betterproto.string_field(1)
expires: datetime = betterproto.message_field(2)
[docs]@dataclass(eq=False, repr=False)
class EnterpriseConfig(betterproto.Message):
"""
EnterpriseConfig is the configuration we store for heartbeating to the
license server.
"""
# license_server is the address of the grpc license service
license_server: str = betterproto.string_field(1)
# id is the unique identifier for this pachd, which is registered with the
# license service
id: str = betterproto.string_field(2)
# secret is a shared secret between this pachd and the license service
secret: str = betterproto.string_field(3)
[docs]@dataclass(eq=False, repr=False)
class EnterpriseRecord(betterproto.Message):
"""
EnterpriseRecord is a protobuf we cache in etcd to store the enterprise
status.
"""
# license is the cached LicenseRecord retrieved from the most recent
# heartbeat to the license server.
license: "LicenseRecord" = betterproto.message_field(1)
# last_heartbeat is the timestamp of the last successful heartbeat to the
# license server
last_heartbeat: datetime = betterproto.message_field(2)
# heartbeat_failed is set if the license is still valid, but the pachd is no
# longer registered with an enterprise server. This is the same as the
# expired state, where auth is locked but not disabled.
heartbeat_failed: bool = betterproto.bool_field(3)
[docs]@dataclass(eq=False, repr=False)
class TokenInfo(betterproto.Message):
"""
TokenInfo contains information about the currently active enterprise token
"""
# expires indicates when the current token expires (unset if there is no
# current token)
expires: datetime = betterproto.message_field(1)
[docs]@dataclass(eq=False, repr=False)
class ActivateRequest(betterproto.Message):
license_server: str = betterproto.string_field(1)
id: str = betterproto.string_field(2)
secret: str = betterproto.string_field(3)
[docs]@dataclass(eq=False, repr=False)
class ActivateResponse(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class GetStateRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class GetStateResponse(betterproto.Message):
state: "State" = betterproto.enum_field(1)
info: "TokenInfo" = betterproto.message_field(2)
# activation_code will always be an empty string, call GetEnterpriseCode to
# get the activation code
activation_code: str = betterproto.string_field(3)
[docs]@dataclass(eq=False, repr=False)
class GetActivationCodeRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class GetActivationCodeResponse(betterproto.Message):
state: "State" = betterproto.enum_field(1)
info: "TokenInfo" = betterproto.message_field(2)
activation_code: str = betterproto.string_field(3)
[docs]@dataclass(eq=False, repr=False)
class HeartbeatRequest(betterproto.Message):
"""
Heartbeat in the enterprise service just triggers a heartbeat for testing
purposes. The RPC used to communicate with the license service is defined
in the license service.
"""
pass
[docs]@dataclass(eq=False, repr=False)
class HeartbeatResponse(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class DeactivateRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class DeactivateResponse(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class PauseRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class PauseResponse(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class UnpauseRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class UnpauseResponse(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class PauseStatusRequest(betterproto.Message):
pass
[docs]@dataclass(eq=False, repr=False)
class PauseStatusResponse(betterproto.Message):
status: "PauseStatusResponsePauseStatus" = betterproto.enum_field(1)
[docs]class ApiStub(betterproto.ServiceStub):
[docs] async def activate(
self, *, license_server: str = "", id: str = "", secret: str = ""
) -> "ActivateResponse":
request = ActivateRequest()
request.license_server = license_server
request.id = id
request.secret = secret
return await self._unary_unary(
"/enterprise_v2.API/Activate", request, ActivateResponse
)
[docs] async def get_state(self) -> "GetStateResponse":
request = GetStateRequest()
return await self._unary_unary(
"/enterprise_v2.API/GetState", request, GetStateResponse
)
[docs] async def get_activation_code(self) -> "GetActivationCodeResponse":
request = GetActivationCodeRequest()
return await self._unary_unary(
"/enterprise_v2.API/GetActivationCode", request, GetActivationCodeResponse
)
[docs] async def heartbeat(self) -> "HeartbeatResponse":
request = HeartbeatRequest()
return await self._unary_unary(
"/enterprise_v2.API/Heartbeat", request, HeartbeatResponse
)
[docs] async def deactivate(self) -> "DeactivateResponse":
request = DeactivateRequest()
return await self._unary_unary(
"/enterprise_v2.API/Deactivate", request, DeactivateResponse
)
[docs] async def pause(self) -> "PauseResponse":
request = PauseRequest()
return await self._unary_unary(
"/enterprise_v2.API/Pause", request, PauseResponse
)
[docs] async def unpause(self) -> "UnpauseResponse":
request = UnpauseRequest()
return await self._unary_unary(
"/enterprise_v2.API/Unpause", request, UnpauseResponse
)
[docs] async def pause_status(self) -> "PauseStatusResponse":
request = PauseStatusRequest()
return await self._unary_unary(
"/enterprise_v2.API/PauseStatus", request, PauseStatusResponse
)
[docs]class ApiBase(ServiceBase):
[docs] async def activate(
self, license_server: str, id: str, secret: str
) -> "ActivateResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def get_state(self) -> "GetStateResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def get_activation_code(self) -> "GetActivationCodeResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def heartbeat(self) -> "HeartbeatResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def deactivate(self) -> "DeactivateResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def pause(self) -> "PauseResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def unpause(self) -> "UnpauseResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs] async def pause_status(self) -> "PauseStatusResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_activate(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"license_server": request.license_server,
"id": request.id,
"secret": request.secret,
}
response = await self.activate(**request_kwargs)
await stream.send_message(response)
async def __rpc_get_state(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.get_state(**request_kwargs)
await stream.send_message(response)
async def __rpc_get_activation_code(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.get_activation_code(**request_kwargs)
await stream.send_message(response)
async def __rpc_heartbeat(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.heartbeat(**request_kwargs)
await stream.send_message(response)
async def __rpc_deactivate(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.deactivate(**request_kwargs)
await stream.send_message(response)
async def __rpc_pause(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.pause(**request_kwargs)
await stream.send_message(response)
async def __rpc_unpause(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.unpause(**request_kwargs)
await stream.send_message(response)
async def __rpc_pause_status(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {}
response = await self.pause_status(**request_kwargs)
await stream.send_message(response)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/enterprise_v2.API/Activate": grpclib.const.Handler(
self.__rpc_activate,
grpclib.const.Cardinality.UNARY_UNARY,
ActivateRequest,
ActivateResponse,
),
"/enterprise_v2.API/GetState": grpclib.const.Handler(
self.__rpc_get_state,
grpclib.const.Cardinality.UNARY_UNARY,
GetStateRequest,
GetStateResponse,
),
"/enterprise_v2.API/GetActivationCode": grpclib.const.Handler(
self.__rpc_get_activation_code,
grpclib.const.Cardinality.UNARY_UNARY,
GetActivationCodeRequest,
GetActivationCodeResponse,
),
"/enterprise_v2.API/Heartbeat": grpclib.const.Handler(
self.__rpc_heartbeat,
grpclib.const.Cardinality.UNARY_UNARY,
HeartbeatRequest,
HeartbeatResponse,
),
"/enterprise_v2.API/Deactivate": grpclib.const.Handler(
self.__rpc_deactivate,
grpclib.const.Cardinality.UNARY_UNARY,
DeactivateRequest,
DeactivateResponse,
),
"/enterprise_v2.API/Pause": grpclib.const.Handler(
self.__rpc_pause,
grpclib.const.Cardinality.UNARY_UNARY,
PauseRequest,
PauseResponse,
),
"/enterprise_v2.API/Unpause": grpclib.const.Handler(
self.__rpc_unpause,
grpclib.const.Cardinality.UNARY_UNARY,
UnpauseRequest,
UnpauseResponse,
),
"/enterprise_v2.API/PauseStatus": grpclib.const.Handler(
self.__rpc_pause_status,
grpclib.const.Cardinality.UNARY_UNARY,
PauseStatusRequest,
PauseStatusResponse,
),
}