caikit.runtime.utils.servicer_util

A generic module to help Predict and Train servicers

Attributes

log
NON_PRIMITIVE_TYPES
CAIKIT_STATUS_CODE_TO_DEBUG_ERROR_TYPE

Classes

ServicePackageStreamWrapper

This class wraps up a DataStreamSourceBase derived class so that it can

Functions

validate_caikit_library_class_exists(cdm, class_name)
validate_caikit_library_class_method_exists(...)
build_proto_stream(...)	Returns an iterator that serializes each item in the model's response to protobuf
build_proto_response(→ google.protobuf.message.Message)	Serializes a data model instance into a protobuf message
is_protobuf_primitive_field(obj)	Check whether or not a descriptor object is a protobufs primitive. The
get_metadata(context, key[, required])	Retrieve a value from the gRPC ServicerContext invocation metadata
validate_data_model(service_descriptor)	Validate the Caikit Library Common Data Model (CDM) against a service descriptor
build_caikit_library_request_dict(→ Dict[str, Any])	Build the request kwargs dict.
raise_caikit_runtime_exception(exception)

Module Contents

caikit.runtime.utils.servicer_util.log

caikit.runtime.utils.servicer_util.NON_PRIMITIVE_TYPES

caikit.runtime.utils.servicer_util.CAIKIT_STATUS_CODE_TO_DEBUG_ERROR_TYPE

caikit.runtime.utils.servicer_util.validate_caikit_library_class_exists(cdm, class_name)

caikit.runtime.utils.servicer_util.validate_caikit_library_class_method_exists(caikit_library_class, method_name)

caikit.runtime.utils.servicer_util.build_proto_stream(caikit_library_response: Iterable[caikit.core.data_model.base.DataBase], context: grpc.ServicerContext) → Iterator[google.protobuf.message.Message]

Returns an iterator that serializes each item in the model’s response to protobuf

caikit.runtime.utils.servicer_util.build_proto_response(caikit_library_response: caikit.core.data_model.base.DataBase) → google.protobuf.message.Message

Serializes a data model instance into a protobuf message

caikit.runtime.utils.servicer_util.is_protobuf_primitive_field(obj)

Check whether or not a descriptor object is a protobufs primitive. The full list of descriptors can be found at the link below.

https://developers.google.com/protocol-buffers/docs/reference/cpp/google.protobuf.descriptor

Args:

obj (google.protobufs.pyext._message.FieldDescriptor):

A protoc-compiled FieldDescriptor object describing a field to be passed as input or output in a well-structured way.

Returns:

boolean indicating whether or not a primitive compiled protobuf field was passed.

caikit.runtime.utils.servicer_util.get_metadata(context, key, required=True)

Retrieve a value from the gRPC ServicerContext invocation metadata dictionary with the given key

Args:

context(grpc.ServicerContext): Context object (contains request metadata, etc) key(string): The invocation metadata dictionary key to retrieve required(boolean): Whether the piece of metadata is required to exist

Returns:

value (object): The value, or None if the specified key does not

exist in the invocation metadata

caikit.runtime.utils.servicer_util.validate_data_model(service_descriptor: google.protobuf.descriptor.ServiceDescriptor)

Validate the Caikit Library Common Data Model (CDM) against a service descriptor that defines the RPCs that this class must support at predict/train time. More specifically, this function will iterate over every RPC/method defined in the service descriptor, and for each RPC, will validate that the Caikit Library CDM has classes that correspond with the field types of the incoming message fields, and that those Caikit Library classes have from_proto functions defined. It will further verify that the output message has a corresponding Caikit Library CDM class with a to_proto method defined.

Args:

service_descriptor(google.protobuf.descriptor.ServiceDescriptor):

A protoc-compiled ServiceDescriptor that defines the predict RPCs that will be serviced by this GlobalPredictServicer

Raises:

AttributeError if a discrepancy is found between the RPC service descriptor and the Caikit Library CDM, which will prevent an instance of this class from being instantiated

class caikit.runtime.utils.servicer_util.ServicePackageStreamWrapper(stream: caikit.runtime.service_generation.data_stream_source.DataStreamSourceBase)

Bases: caikit.runtime.service_generation.data_stream_source.DataStreamSourceBase

This class wraps up a DataStreamSourceBase derived class so that it can be safely pickled to a subprocess using “spawn”.

When using “spawn” to start a subprocess, a clean python process is created and none of the runtime code leading to the creation is re-executed. For most classes that are created at import time, this is fine, but for dynamically created classes like those created with make_data_stream_source, the class is not recreated on spawn and cannot be looked up.

To get around this, when unpickling this class, it falls back to performing service generation dynamically if the class doesn’t already exist

_stream

The internal _stream is cached here so that the result of calling to_data_stream can be re-read, rather than requiring to_data_stream to be invoked on every read through the stream

__getstate__() → Tuple[str, str, type, bytes]

Pickle as a tuple of primitive types

Returns:

stream_module_name (str): The name of the module that holds the

stream source class

stream_class_name (str): The name of the stream source class itself stream_class_element_type (type): The type of the elements for this

stream source

content (bytes): The serialized bytes content of the underlying

stream source as a dataobject

__setstate__(pickled: Tuple[str, str, type, bytes])

Unpickle the values from pickling and fall back to dynamic service generation if needed.

Args:

pickled (Tuple[str, str, type, bytes]): The four elements created

when pickling this instance

__getattr__(name: str) → Any

Forward all getattr requests to the underlying stream

__setattr__(name: str, val: Any)

Forward attribute setting for generic attrs to the stream

caikit.runtime.utils.servicer_util.build_caikit_library_request_dict(request: google.protobuf.message.Message, module_signature: caikit.core.signature_parsing.CaikitMethodSignature) → Dict[str, Any]

Build the request kwargs dict.

Args:

request (ProtoMessageType):

The request proto message to deserialize from

module_signature (CaikitMethodSignature):

Module signature or metadata about method on a module

Returns:

kwargs dict

caikit.runtime.utils.servicer_util.raise_caikit_runtime_exception(exception: caikit.core.exceptions.caikit_core_exception.CaikitCoreException)