Source code for mindsphere_core.mindsphere_core

# -*- coding: utf-8 -*-
    This module implements the common features of sdk used by all clients.

    Copyright (C), Siemens AG 2018 Licensed under the
    MindSphere Developer License, see

import os
import os.path
from mindsphere_core import log_config
import requests
import json
from configparser import ConfigParser, ExtendedInterpolation

from . import commonutil
from . import constants
from . import exceptions
from . import serialization_filter
from . import error_messages
from . import token_service

logger = log_config.default_logging()

path = os.path.abspath(__file__)
config_file = os.path.dirname(path) + r"/url_setup.ini"

[docs]class RestClientConfig: """This class allows to configure proxy , timeout and hosting environment.""" def __init__( self, proxy_host=None, proxy_port=None, proxy_username=None, proxy_password=None, connection_timeout_in_seconds=constants.TIME_OUT_DEFAULT, socket_timeout_in_seconds=constants.TIME_OUT_DEFAULT, proxy_schema=constants.PROXY_SCHEMA_DEFAULT, host_environment=None, ): """ proxy_host : str Proxy host proxy_username : str Proxy User name proxy_port : str Proxy port proxy_password : str Proxy password connection_timeout_in_seconds : str Connection timeout socket_timeout_in_seconds : str Socket timeout proxy_schema : str Proxy schema host_environment : str Target hosted environment """ self.proxy_host = proxy_host self.proxy_port = proxy_port self.proxy_username = proxy_username self.proxy_password = proxy_password self.connection_timeout_in_seconds = connection_timeout_in_seconds self.socket_timeout_in_seconds = socket_timeout_in_seconds self.proxy_schema = proxy_schema self.host_environment = host_environment
[docs]def fetch_token(rest_client_config, credentials): """Fetches a token either using service credentials or validates the token if its already provided in credentials """ return token_service.fetch_token(rest_client_config, credentials)
[docs]def build_url(base_path, end_point_url, config=None): """builds the end point url based on host environment. base_path : str base path for the selected service end_point_url : str url for the requested service end point config : RestClientConfig config object set by user """ if config and config.host_environment: host_environment = config.host_environment elif constants.HOST_ENVIRONMENT in os.environ: host_environment = os.environ[constants.HOST_ENVIRONMENT] else: host_environment = constants.HOST_ENV_DEFAULT parser = ConfigParser(interpolation=ExtendedInterpolation()) service_url_pre = parser.get(constants.SERVICE_URL, constants.SERVICE_URL_PRE) service_url_post = parser.get(constants.SERVICE_URL, constants.SERVICE_URL_POST) return "".join( [service_url_pre, host_environment, service_url_post, base_path, end_point_url] )
[docs]def invoke_service( rest_client_config, api_url, headers, http_method, query_params=None, form_params=None, body_params=None, local_var_files=None, response_type=None, model_package=None, ): """ Method to invoke rest end points. """ if rest_client_config and rest_client_config.connection_timeout_in_seconds: timeout = float(rest_client_config.connection_timeout_in_seconds) else: timeout = None if ( rest_client_config and rest_client_config.proxy_host and rest_client_config.proxy_port ): http_proxy = ( "http://" + rest_client_config.proxy_host + ":" + rest_client_config.proxy_port ) proxy_dict = {"https": http_proxy} else: proxy_dict = None if headers: headers = serialization_filter.sanitize_headers(headers) if body_params: # Serialize the input object payload = serialization_filter.sanitize_for_serialization(body_params) if type(payload) is not bytes: # Converts object to string if object # not of type bytes payload = json.dumps(payload) elif form_params: payload = form_params else: payload = None try: response = requests.request( http_method, api_url, params=query_params, data=payload, files=local_var_files, headers=headers, proxies=proxy_dict, timeout=timeout, ) response.raise_for_status() if response.text: json_data = response.text if not response_type or response_type != constants.STR_TYPE: # Converts response to object # Except when response type is string json_data = json.loads(response.text) if response_type: # Finds the relative path to model package serialization_filter.service_model_package = ( model_package + constants.MODELS_PACKAGE ) # De-serialize the response to desired response type json_data = serialization_filter.deserialize(json_data, response_type) return json_data except requests.exceptions.HTTPError as error_http: error = commonutil.parse_exception(error_http.response) raise exceptions.MindsphereServerError(error) except requests.exceptions.ConnectionError as error_connection: raise exceptions.MindsphereServerError( error_messages.CONNECTION_ERROR, error_connection ) except requests.exceptions.Timeout as error_time_out: raise exceptions.MindsphereServerError( error_messages.TIMEOUT_ERROR, error_time_out ) except requests.exceptions.RequestException as error: raise exceptions.MindsphereServerError( error_messages.UN_IDENTIFIED_ERROR, error )