# 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.
"""Mock record copier class definition."""
from typing import Dict, TYPE_CHECKING
if TYPE_CHECKING:
from .record_creator import RecordCreator
from .record_info import RecordInfo
from .record_ref import RecordRef
[docs]class RecordCopier:
"""Record copier mock."""
def __init__(
self,
destination: "RecordInfo",
source: "RecordInfo",
suppress_size_only_conversion_errors: bool = False,
decimal_separator: str = ".",
) -> None:
"""Construct a record copier."""
self.destination = destination
self.source = source
self.supress_size_only_conversion_errors = suppress_size_only_conversion_errors
self.decimal_separator = decimal_separator
self._done_adding = False
self._idx_map: Dict[int, int] = {}
[docs] def add(self, destination_field_num: int, source_field_num: int) -> None:
"""Add a field to the record copier."""
if self._done_adding:
raise RuntimeError("Done adding, can't add any more.")
self._idx_map[source_field_num] = destination_field_num
[docs] def done_adding(self) -> None:
"""Signal that fields are done being added."""
self._done_adding = True
[docs] def copy(self, destination: "RecordCreator", source: "RecordRef") -> None:
"""Copy a record to the destination format."""
for source_idx, dest_idx in self._idx_map.items():
source_key = list(source.data.keys())[source_idx]
dest_key = list(destination.record_ref.data.keys())[dest_idx]
destination.record_ref.data[dest_key] = source.data[source_key]
[docs] @staticmethod
def set_dest_to_null(destination: "RecordCreator") -> None:
"""Set destination to all null."""
for key in destination.record_ref.data:
destination.record_ref.data[key] = None