Source code for ayx_python_sdk.providers.amp_provider.grpc_helpers.dcme_msgs

# 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.
"""AMP Provider: Proxy class for DCM API."""
import datetime as dt
from typing import Dict, Optional

from ayx_python_sdk.core.exceptions import DcmEError, DcmEException
from ayx_python_sdk.providers.amp_provider.resources.generated.dcm_e_pb2 import (
    DcmERequest,
    DcmEResponse,
)

from google.protobuf import json_format
from google.protobuf.struct_pb2 import Struct

from .control_msgs import DCME_REQ, ctrl_out_wrapper


[docs]@ctrl_out_wrapper(DCME_REQ, set_id=True) def get_connection_msg(connection_id: str) -> DcmERequest: # noqa: D102 """Create a connection request message.""" req = DcmERequest() req.v2.get_connection.connection_id = connection_id return req
[docs]@ctrl_out_wrapper(DCME_REQ) def get_write_lock_msg( connection_id: str, role: str, secret_type: str, expires_in: Optional[dt.datetime] = None, ) -> DcmERequest: # noqa: D102 """Create a LockSecret message.""" req = DcmERequest() req.v2.lock_secret.connection_id = connection_id req.v2.lock_secret.credential_role = role req.v2.lock_secret.secret_type = secret_type if expires_in: req.v2.lock_secret.expires_in = expires_in.isoformat() return req
[docs]@ctrl_out_wrapper(DCME_REQ) def free_write_lock_msg( connection_id: str, role: str, secret_type: str, lock_id: str ) -> DcmERequest: # noqa: D102 """Create a UnlockSecret message.""" req = DcmERequest() req.v2.unlock_secret.connection_id = connection_id req.v2.unlock_secret.credential_role = role req.v2.unlock_secret.secret_type = secret_type req.v2.unlock_secret.lock_id = lock_id return req
[docs]@ctrl_out_wrapper(DCME_REQ) def update_connection_secret_msg( connection_id: str, lock_id: str, role: str, secret_type: str, value: str, expires_on: Optional[dt.datetime], parameters: Optional[Dict[str, str]], ) -> DcmERequest: # noqa: D102 """Create UpdateSecret message.""" req = DcmERequest() req.v2.update_secret.connection_id = connection_id req.v2.update_secret.lock_id = lock_id req.v2.update_secret.credential_role = role req.v2.update_secret.secret_type = secret_type req.v2.update_secret.value = value if parameters: st = Struct() st.update(parameters) req.v2.update_secret.parameters.CopyFrom(st) if expires_on: req.v2.update_secret.expires_on = expires_on.isoformat() return req
[docs]def handle_response(res: DcmEResponse) -> Dict: """Handle DcmE response.""" response_dict = json_format.MessageToDict(res.response) if not res.success: if "message" in response_dict.keys() and "errorCode" in response_dict.keys(): if "detail" in response_dict.keys(): response_dict["dcme_exception"] = DcmEException( "Dcm.E Error!", DcmEError( response_dict["message"], int(response_dict["errorCode"]), response_dict["detail"], ), ) else: response_dict["dcme_exception"] = DcmEException( "Dcm.E Error!", DcmEError( response_dict["message"], int(response_dict["errorCode"]) ), ) else: response_dict["dcme_exception"] = DcmEException( "Dcm.E Error!", DcmEError(res.response, 10003) ) return response_dict