Source code for ayx_python_sdk.providers.amp_provider.repositories.io_repository

# 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.
"""Class that saves and retrieves AMP IO information."""
import logging
import uuid
from pathlib import Path
from typing import Any, cast

from ayx_python_sdk.core.exceptions import WorkflowRuntimeError
from ayx_python_sdk.providers.amp_provider.repositories import EnvironmentRepository
from ayx_python_sdk.providers.amp_provider.repositories import GrpcRepository
from ayx_python_sdk.providers.amp_provider.repositories.singleton import Singleton
from ayx_python_sdk.providers.amp_provider.resources.generated.output_message_data_pb2 import (
    OutputMessageData,
    OutputMessageTypes,
)
from ayx_python_sdk.providers.amp_provider.resources.generated.password_data_pb2 import (
    PasswordData,
)
from ayx_python_sdk.providers.amp_provider.resources.generated.translate_message_data_pb2 import (
    TranslateMessageData,
)

logger = logging.getLogger(__name__)


[docs]class IORepository(metaclass=Singleton): """Repository that stores IO information.""" def __init__(self) -> None: """Initialize the IO repository."""
[docs] @staticmethod def save_error(error_msg: str) -> None: """ Display an error message in the Results window. Parameters ---------- error_msg The error message to be output. """ GrpcRepository().get_sdk_engine_client().OutputMessage( OutputMessageData( message_type=OutputMessageTypes.OMT_Error, message=error_msg ) )
[docs] @staticmethod def save_warn(warn_msg: str) -> None: """ Display a warning message in the Results window. Parameters ---------- warn_msg The warning message to be output. """ GrpcRepository().get_sdk_engine_client().OutputMessage( OutputMessageData( message_type=OutputMessageTypes.OMT_Warning, message=warn_msg ) )
[docs] @staticmethod def save_info(info_msg: str) -> None: """ Display an info message in the Results window. Parameters ---------- info_msg The info message to be output. """ GrpcRepository().get_sdk_engine_client().OutputMessage( OutputMessageData( message_type=OutputMessageTypes.OMT_Info, message=info_msg ) )
[docs] @staticmethod def get_translate_msg(msg: str, *args: Any) -> str: """ Translate a message to the current locale. Parameters ---------- msg Message to translate. *args Interpolation data for the string. Returns ------- str The message, translated into the locale. """ interpolation_data = [str(arg) for arg in args] if len(args) > 0 else [] res = ( GrpcRepository() .get_sdk_engine_client() .TranslateMessage( TranslateMessageData( unlocalized_string=msg, interpolation_data=interpolation_data ) ) ) return cast(str, res.translated_message)
[docs] @staticmethod def update_progress(percent: float) -> None: """ Update tool progress. Parameters ---------- percent The progress (how close the connection is to processing all records) as a percentage. """
# TODO
[docs] @staticmethod def get_temp_file(extension: str = "tmp") -> "Path": """ Create a temporary file managed by Designer. Parameters ---------- extension The file extension of the temp file. Returns ------- Path The path to where the temp file is. """ temp_file_name = f"temp-file-{str(uuid.uuid4())}.{str(extension)}" engine_temp_dir = EnvironmentRepository().get_temp_dir() temp_file_path = Path(engine_temp_dir) / (temp_file_name) try: temp_file_path.touch() except FileNotFoundError: # path does not exist logger.error("Engine.TempFilePath (%s) does not exist", engine_temp_dir) except IOError: # path exists but no write permissions logger.error("No write permissions for directory %s", engine_temp_dir) return Path(temp_file_path)
[docs] @staticmethod def decrypt_password(password: str) -> str: # noqa: D102 """ Decrypt password. Parameters ---------- password The password to decrypt. Returns ------- str The decrypted password. """ import grpc try: decrypted = ( GrpcRepository() .get_sdk_engine_client() .DecryptPassword(PasswordData(password=password)) ) except grpc.RpcError: raise WorkflowRuntimeError("Error decrypting password") return cast(str, decrypted.password)