# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import ca_pb2 as ca__pb2
GRPC_GENERATED_VERSION = '1.67.1'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ f' but the generated code in ca_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
[docs]
class CertificateAuthorityStub(object):
"""Missing associated documentation comment in .proto file."""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.Enroll = channel.unary_unary(
'/CertificateAuthority/Enroll',
request_serializer=ca__pb2.CSR.SerializeToString,
response_deserializer=ca__pb2.CRT.FromString,
_registered_method=True)
self.Reenroll = channel.unary_unary(
'/CertificateAuthority/Reenroll',
request_serializer=ca__pb2.CSR.SerializeToString,
response_deserializer=ca__pb2.CRT.FromString,
_registered_method=True)
[docs]
class CertificateAuthorityServicer(object):
"""Missing associated documentation comment in .proto file."""
[docs]
def Enroll(self, request, context):
"""Enroll is called by client passing Certificate Signing Request (CSR)
containing its personal certificate to be signed. The server returns
back the certificate (CRT) to authenticate into IngestGateway or Router.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
[docs]
def Reenroll(self, request, context):
"""Reenroll is called by client passing Certificate Signing Request (CSR)
when it wants to renew its personal certificate. The server returns
a new certificate (CRT).
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
[docs]
def add_CertificateAuthorityServicer_to_server(servicer, server):
rpc_method_handlers = {
'Enroll': grpc.unary_unary_rpc_method_handler(
servicer.Enroll,
request_deserializer=ca__pb2.CSR.FromString,
response_serializer=ca__pb2.CRT.SerializeToString,
),
'Reenroll': grpc.unary_unary_rpc_method_handler(
servicer.Reenroll,
request_deserializer=ca__pb2.CSR.FromString,
response_serializer=ca__pb2.CRT.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'CertificateAuthority', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('CertificateAuthority', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
[docs]
class CertificateAuthority(object):
"""Missing associated documentation comment in .proto file."""
[docs]
@staticmethod
def Enroll(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/CertificateAuthority/Enroll',
ca__pb2.CSR.SerializeToString,
ca__pb2.CRT.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
[docs]
@staticmethod
def Reenroll(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/CertificateAuthority/Reenroll',
ca__pb2.CSR.SerializeToString,
ca__pb2.CRT.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)