caikit.core.data_model.streams.converter

DataStream converter that converts a given Datastream to either a DataStream of lists or dictionaries. These type conversions are encapsulated here right now for consistency in the workflows served in production cloud offerings, but we could see moving these to be directly on the DataStream class.

Attributes

log
error

Classes

DataStreamConverter

Converts DataStreams to a target type

Module Contents

caikit.core.data_model.streams.converter.log

caikit.core.data_model.streams.converter.error

class caikit.core.data_model.streams.converter.DataStreamConverter(target_type: type, key_list: List[str])

Converts DataStreams to a target type

This uses a target type and a list of expected keys to convert each data item in the stream.

For example, for an input stream that looks like:

[

{ a: 1, b: 2, c: 3 } { a: 4, b: 5, c: 6 } { a: 7, b: 8, c: 9 }

]

If target_type is list with key_list of [‘a’,’c’], the result will be:

[

[1, 3] [4, 6] [7, 9]

]

Or, for an input stream that looks like:

[

[1, 3] [4, 6] [7, 9]

]

If target_type is dict with key_list of [‘foo’, ‘bar’], the result will be:

[

{ foo: 1, bar: 3 } { foo: 4, bar: 6 } { foo: 7, bar: 9 }

]

>>> list_of_dicts = [{"foo": 1, "bar": 2}, {"foo": 3, "bar": 4}]
>>> dict_stream = DataStream.from_iterable(list_of_dicts)
>>> converter = DataStreamConverter(target_type=list, key_list=['foo', 'bar'])
>>> list_stream = converter.convert(dict_stream)

_conversion_function_map: Dict[type, Callable]

_target_type: type

_key_list: List[str]

convert(stream: caikit.core.data_model.DataStream) → caikit.core.data_model.DataStream

Attempt to convert a given datastream to a datastream of the target type See classdoc for examples

Args:

stream (DataStream): stream intended to be converted

Returns:

Converted datastream based on the target type

_convert_stream_to_dicts(stream: caikit.core.data_model.DataStream) → caikit.core.data_model.DataStream

Attempt to convert a stream to dictionaries

Args:

stream (DataStream): Stream to convert to a stream of dictionaries

Returns:

A stream which will lazily convert data items to dictionaries

_convert_stream_to_lists(stream: caikit.core.data_model.DataStream) → caikit.core.data_model.DataStream

Attempt to convert a stream to lists

Args:

stream (DataStream): Stream to convert to a stream of lists

Returns:

A stream which will lazily convert data items to lists