Source code for ayx_python_sdk.providers.amp_provider.builders.record_builder

# Copyright (C) 2022 Alteryx, Inc. All rights reserved.
#
# Licensed under the ALTERYX SDK AND API LICENSE AGREEMENT;
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    https://www.alteryx.com/alteryx-sdk-and-api-license-agreement
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Record building utilities for converting between core and protobuf."""
from typing import Any, Dict, List, Sequence

from ayx_python_sdk.core.constants import NULL_VALUE_PLACEHOLDER
from ayx_python_sdk.core.field import FieldType
from ayx_python_sdk.core.metadata import Metadata
from ayx_python_sdk.providers.amp_provider.builders.packers import (
    _BlobPacker,
    _BoolFalsePacker,
    _BoolPacker,
    _BoolTruePacker,
    _BytePacker,
    _DatePacker,
    _DatetimePacker,
    _DoublePacker,
    _EmptyStringPacker,
    _FloatPacker,
    _IndirectBlobPacker,
    _IndirectStringPacker,
    _Int0Packer,
    _Int16Packer,
    _IntegerPacker,
    _LongIntegerPacker,
    _NullPacker,
    _Packer,
    _StringPacker,
)
from ayx_python_sdk.providers.amp_provider.resources.generated.record_packet_pb2 import (
    AMPFieldType,
    Record as ProtobufRecord,
)


[docs]class RecordBuilder: """Utilities for converting records between protobuf and core format objects.""" _field_type_to_packer: Dict[Any, _Packer] = { AMPFieldType.STRING: _StringPacker(), AMPFieldType.BLOB: _BlobPacker(), AMPFieldType.SPATIALOBJ: _BlobPacker(), AMPFieldType.BCD: _BlobPacker(), AMPFieldType.BOOL: _BoolPacker(), AMPFieldType.INT0: _Int0Packer(), AMPFieldType.UINT8: _BytePacker(), AMPFieldType.INT16: _Int16Packer(), AMPFieldType.INT32: _IntegerPacker(), AMPFieldType.INT64: _LongIntegerPacker(), AMPFieldType.FLOAT: _FloatPacker(), AMPFieldType.DOUBLE: _DoublePacker(), AMPFieldType.DATE: _DatePacker(), AMPFieldType.TIME: _IntegerPacker(), AMPFieldType.DATETIME: _DatetimePacker(), AMPFieldType.INDIRECTSTRING: _IndirectStringPacker(), AMPFieldType.INDIRECTBLOB: _IndirectBlobPacker(), AMPFieldType.INDIRECTSPATIAL: _IndirectBlobPacker(), AMPFieldType.BOOLFALSE: _BoolFalsePacker(), AMPFieldType.BOOLTRUE: _BoolTruePacker(), AMPFieldType.EMPTYSTRING: _EmptyStringPacker(), AMPFieldType.NULL: _NullPacker(), } _blob_cosmetic_types_to_amp_type = { FieldType.blob: AMPFieldType.BLOB, FieldType.spatialobj: AMPFieldType.SPATIALOBJ, FieldType.fixeddecimal: AMPFieldType.BCD, } _datetime_cosmetic_types_to_amp_type = { FieldType.date: AMPFieldType.DATE, FieldType.datetime: AMPFieldType.DATETIME, FieldType.time: AMPFieldType.TIME, } _float_cosmetic_types_to_amp_type = { FieldType.float: AMPFieldType.FLOAT, FieldType.double: AMPFieldType.DOUBLE, } _int_cosmetic_types_to_amp_type = { FieldType.byte: AMPFieldType.UINT8, FieldType.int16: AMPFieldType.INT16, FieldType.int32: AMPFieldType.INT32, FieldType.int64: AMPFieldType.INT64, } _null_values = {NULL_VALUE_PLACEHOLDER, None}
[docs] @classmethod def from_protobuf(cls, record: ProtobufRecord) -> List: """ Convert a protobuf to a list of values. Parameters ---------- record Protobuf representation of a record. Returns ------- List All elements of the record as a list. """ start_byte_idx = 0 parsed_record = [] for amp_type in record.types: try: packer = cls._field_type_to_packer[amp_type] except KeyError: raise ValueError(f"Packer not found for type: {amp_type}") element, stride = packer.unpack(record.data, start_byte_idx) parsed_record.append(element) start_byte_idx += stride return parsed_record
[docs] @classmethod def to_protobuf(cls, record: Sequence, metadata: Metadata) -> ProtobufRecord: """ Convert a sequence of values to a protobuf. Parameters ---------- record A sequence of values. metadata Metadata associated with the values. Returns ------- ProtobufRecord The protobuf representation of the passed in record. """ assert len(record) == len(metadata) raw_data = bytes() amp_types = [] for field, element in zip(metadata, record): try: amp_type = cls._get_amp_type(field.type, element) packer = cls._field_type_to_packer[amp_type] except KeyError: raise ValueError(f"Packer not found for type: {field.type}") raw_data += packer.pack(element) amp_types.append(amp_type) return ProtobufRecord(data=raw_data, types=amp_types)
@classmethod def _get_amp_type(cls, cosmetic_type: FieldType, element: Any) -> AMPFieldType: if element in cls._null_values: return AMPFieldType.NULL if cosmetic_type in [ FieldType.string, FieldType.v_string, FieldType.v_wstring, FieldType.wstring, ]: if element == "": return AMPFieldType.EMPTYSTRING return AMPFieldType.STRING if cosmetic_type in cls._blob_cosmetic_types_to_amp_type: return cls._blob_cosmetic_types_to_amp_type[cosmetic_type] if cosmetic_type == FieldType.bool: if element: return AMPFieldType.BOOLTRUE return AMPFieldType.BOOLFALSE if element == 0: return AMPFieldType.INT0 if cosmetic_type in cls._float_cosmetic_types_to_amp_type: return cls._float_cosmetic_types_to_amp_type[cosmetic_type] if cosmetic_type in cls._datetime_cosmetic_types_to_amp_type: return cls._datetime_cosmetic_types_to_amp_type[cosmetic_type] if cosmetic_type in cls._int_cosmetic_types_to_amp_type: return cls._int_cosmetic_types_to_amp_type[cosmetic_type] raise ValueError( f"AMP field type not found for cosmetic type {cosmetic_type} and value {element}." )