# Unless explicitly stated otherwise all files in this repository are licensed under the Apache-2.0 License. # This product includes software developed at Datadog (https://www.datadoghq.com/). # Copyright 2019-Present Datadog, Inc. import copy import logging import os import urllib3 # type: ignore from http import client as http_client from datadog_api_client.exceptions import ApiValueError JSON_SCHEMA_VALIDATION_KEYWORDS = { "multipleOf", "maximum", "exclusiveMaximum", "minimum", "exclusiveMinimum", "maxLength", "minLength", "pattern", "maxItems", "minItems", } class _UnstableOperations: def __init__(self, values): self.values = values def get(self, key, default=None): if key in self: return self[key] return default def __getitem__(self, key): if key in self.values: return self.values[key] for version in ("v1", "v2"): version_key = f"{version}.{key}" if version_key in self.values: return self.values[version_key] raise KeyError(f"Unknown unstable operation {key}") def __setitem__(self, key, value): if key in self.values: self.values[key] = value for version in ("v1", "v2"): version_key = f"{version}.{key}" if version_key in self.values: self.values[version_key] = value break else: raise KeyError(f"Unknown unstable operation {key}") def __contains__(self, key): if key in self.values: return True for version in ("v1", "v2"): version_key = f"{version}.{key}" if version_key in self.values: return True return False class Configuration: """ :param host: Base url. :param api_key: Dict to store API key(s). Each entry in the dict specifies an API key. The dict key is the name of the security scheme in the OAS specification. The dict value is the API key secret. :param api_key_prefix: Dict to store API prefix (e.g. Bearer). The dict key is the name of the security scheme in the OAS specification. The dict value is an API key prefix when generating the auth data. :param username: Username for HTTP basic authentication. :param password: Password for HTTP basic authentication. :param discard_unknown_keys: Boolean value indicating whether to discard unknown properties. A server may send a response that includes additional properties that are not known by the client in the following scenarios: 1. The OpenAPI document is incomplete, i.e. it does not match the server implementation. 2. The client was generated using an older version of the OpenAPI document and the server has been upgraded since then. If a schema in the OpenAPI document defines the additionalProperties attribute, then all undeclared properties received by the server are injected into the additional properties map. In that case, there are undeclared properties, and nothing to discard. :param disabled_client_side_validations: Comma-separated list of JSON schema validation keywords to disable JSON schema structural validation rules. The following keywords may be specified: multipleOf, maximum, exclusiveMaximum, minimum, exclusiveMinimum, maxLength, minLength, pattern, maxItems, minItems. By default, the validation is performed for data generated locally by the client and data received from the server, independent of any validation performed by the server side. If the input data does not satisfy the JSON schema validation rules specified in the OpenAPI document, an exception is raised. If disabled_client_side_validations is set, structural validation is disabled. This can be useful to troubleshoot data validation problem, such as when the OpenAPI document validation rules do not match the actual API data received by the server. :type disabled_client_side_validations: str :param server_index: Index to servers configuration. :param server_variables: Mapping with string values to replace variables in templated server configuration. The validation of enums is performed for variables with defined enum values before. :param server_operation_index: Mapping from operation ID to an index to server configuration. :param server_operation_variables: Mapping from operation ID to a mapping with string values to replace variables in templated server configuration. The validation of enums is performed for variables with defined enum values before. :param ssl_ca_cert: The path to a file of concatenated CA certificates in PEM format. :param compress: Boolean indicating whether encoded responses are accepted or not. :type compress: bool :param return_http_data_only: Response data without head status code and headers. Default is True. :type return_http_data_only: bool :param preload_content: If False, the urllib3.HTTPResponse object will be returned without reading/decoding response data. Default is True. :type preload_content: bool :param request_timeout: Timeout setting for this request. If one number is provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. Default is None. :type request_timeout: float/tuple :param check_input_type: Specifies if type checking should be done on the data sent to the server. Default is True. :type check_input_type: bool :param check_return_type: Specifies if type checking should be done on the data received from the server. Default is True. :type check_return_type: bool :param spec_property_naming: Whether names in properties are expected to respect the spec or use snake case. :type spec_property_naming: bool :param enable_retry: If set, the client will retry requests on backend errors (5xx status codes), and 429. On 429 if will use the returned headers to wait until the next requests, otherwise it will retry using the backoff factor. :type enable_retry: bool :param retry_backoff_factor: Factor used to space out retried requests on backend errors. :type retry_backoff_factor: float :param max_retries: The maximum number of times a single request can be retried. :type max_retries: int """ def __init__( self, host=None, api_key=None, api_key_prefix=None, access_token=None, username=None, password=None, discard_unknown_keys=True, disabled_client_side_validations="", server_index=None, server_variables=None, server_operation_index=None, server_operation_variables=None, ssl_ca_cert=None, compress=True, return_http_data_only=True, preload_content=True, request_timeout=None, check_input_type=True, check_return_type=True, spec_property_naming=False, enable_retry=False, retry_backoff_factor=2, max_retries=3, ): """Constructor.""" self._base_path = "https://api.datadoghq.com" if host is None else host self.server_index = 0 if server_index is None and host is None else server_index self.server_operation_index = server_operation_index or {} self.server_variables = server_variables or {} self.server_operation_variables = server_operation_variables or {} self.temp_folder_path = None # Authentication Settings self.access_token = access_token self.api_key = {} if api_key: self.api_key = api_key self.api_key_prefix = {} if api_key_prefix: self.api_key_prefix = api_key_prefix self.refresh_api_key_hook = None self.username = username self.password = password self.discard_unknown_keys = discard_unknown_keys self.disabled_client_side_validations = disabled_client_side_validations self.logger = {} self.logger["package_logger"] = logging.getLogger("datadog_api_client") self.logger["urllib3_logger"] = logging.getLogger("urllib3") self.logger_format = "%(asctime)s %(levelname)s %(message)s" self.logger_stream_handler = None self.logger_file_handler = None self.logger_file = None self.debug = False self.verify_ssl = True self.ssl_ca_cert = ssl_ca_cert self.cert_file = None self.key_file = None self.assert_hostname = None self.proxy = None self.proxy_headers = None self.safe_chars_for_path_param = "" # Enable client side validation self.client_side_validation = True # Options to pass down to the underlying urllib3 socket self.socket_options = None # Will translate to a Accept-Encoding header self.compress = compress self.return_http_data_only = return_http_data_only self.preload_content = preload_content self.request_timeout = request_timeout self.check_input_type = check_input_type self.check_return_type = check_return_type self.spec_property_naming = spec_property_naming # Options for http retry self.enable_retry = enable_retry self.retry_backoff_factor = retry_backoff_factor self.max_retries = max_retries # Keep track of unstable operations self.unstable_operations = _UnstableOperations( { "v2.create_open_api": False, "v2.delete_open_api": False, "v2.get_open_api": False, "v2.list_apis": False, "v2.update_open_api": False, "v2.create_app": False, "v2.delete_app": False, "v2.delete_apps": False, "v2.get_app": False, "v2.list_apps": False, "v2.update_app": False, "v2.deploy_app": False, "v2.disable_app": False, "v2.get_active_billing_dimensions": False, "v2.get_billing_dimension_mapping": False, "v2.get_monthly_cost_attribution": False, "v2.cancel_data_deletion_request": False, "v2.create_data_deletion_request": False, "v2.get_data_deletion_requests": False, "v2.create_dora_deployment": False, "v2.create_dora_incident": False, "v2.create_incident": False, "v2.create_incident_integration": False, "v2.create_incident_todo": False, "v2.create_incident_type": False, "v2.delete_incident": False, "v2.delete_incident_integration": False, "v2.delete_incident_todo": False, "v2.delete_incident_type": False, "v2.get_incident": False, "v2.get_incident_integration": False, "v2.get_incident_todo": False, "v2.get_incident_type": False, "v2.list_incident_attachments": False, "v2.list_incident_integrations": False, "v2.list_incidents": False, "v2.list_incident_todos": False, "v2.list_incident_types": False, "v2.search_incidents": False, "v2.update_incident": False, "v2.update_incident_attachments": False, "v2.update_incident_integration": False, "v2.update_incident_todo": False, "v2.update_incident_type": False, "v2.create_aws_account": False, "v2.create_new_aws_external_id": False, "v2.delete_aws_account": False, "v2.get_aws_account": False, "v2.list_aws_accounts": False, "v2.list_aws_namespaces": False, "v2.update_aws_account": False, "v2.list_aws_logs_services": False, "v2.cancel_historical_job": False, "v2.convert_job_result_to_signal": False, "v2.delete_historical_job": False, "v2.get_finding": False, "v2.get_historical_job": False, "v2.list_findings": False, "v2.list_historical_jobs": False, "v2.mute_findings": False, "v2.run_historical_job": False, "v2.create_scorecard_outcomes_batch": False, "v2.create_scorecard_rule": False, "v2.delete_scorecard_rule": False, "v2.list_scorecard_outcomes": False, "v2.list_scorecard_rules": False, "v2.update_scorecard_rule": False, "v2.create_incident_service": False, "v2.delete_incident_service": False, "v2.get_incident_service": False, "v2.list_incident_services": False, "v2.update_incident_service": False, "v2.create_slo_report_job": False, "v2.get_slo_report": False, "v2.get_slo_report_job_status": False, "v2.create_incident_team": False, "v2.delete_incident_team": False, "v2.get_incident_team": False, "v2.list_incident_teams": False, "v2.update_incident_team": False, } ) # Load default values from environment if "DD_SITE" in os.environ: self.server_variables["site"] = os.environ["DD_SITE"] if "DD_API_KEY" in os.environ and not self.api_key.get("apiKeyAuth"): self.api_key["apiKeyAuth"] = os.environ["DD_API_KEY"] if "DD_APP_KEY" in os.environ and not self.api_key.get("appKeyAuth"): self.api_key["appKeyAuth"] = os.environ["DD_APP_KEY"] def __deepcopy__(self, memo): cls = self.__class__ result = cls.__new__(cls) memo[id(self)] = result for k, v in self.__dict__.items(): if k not in ("logger", "logger_file_handler"): setattr(result, k, copy.deepcopy(v, memo)) # Shallow copy of loggers result.logger = copy.copy(self.logger) # Use setters to configure loggers result.logger_file = self.logger_file result.debug = self.debug return result def __setattr__(self, name, value): object.__setattr__(self, name, value) if name == "disabled_client_side_validations": s = set(filter(None, value.split(","))) for v in s: if v not in JSON_SCHEMA_VALIDATION_KEYWORDS: raise ApiValueError("Invalid keyword: '{0}''".format(v)) self._disabled_client_side_validations = s @property def logger_file(self): """The logger file. If the logger_file is None, then add stream handler and remove file handler. Otherwise, add file handler and remove stream handler. :return: The logger_file path. :rtype: str """ return self._logger_file @logger_file.setter def logger_file(self, value): """The logger file. If the logger_file is None, then add stream handler and remove file handler. Otherwise, add file handler and remove stream handler. :param value: The logger_file path. :type value: str """ self._logger_file = value if self._logger_file: # If set logging file, # then add file handler and remove stream handler. self.logger_file_handler = logging.FileHandler(self._logger_file) self.logger_file_handler.setFormatter(self.logger_formatter) for _, logger in self.logger.items(): logger.addHandler(self.logger_file_handler) @property def debug(self): """Debug status. :return: The debug status, True or False. :rtype: bool """ return self._debug @debug.setter def debug(self, value): """Debug status. :param value: The debug status, True or False. :type value: bool """ self._debug = value if self._debug: # if debug status is True, turn on debug logging for _, logger in self.logger.items(): logger.setLevel(logging.DEBUG) # turn on http_client debug http_client.HTTPConnection.debuglevel = 1 else: # if debug status is False, turn off debug logging, # setting log level to default `logging.WARNING` for _, logger in self.logger.items(): logger.setLevel(logging.WARNING) # turn off http_client debug http_client.HTTPConnection.debuglevel = 0 @property def logger_format(self): """The logger format. The logger_formatter will be updated when sets logger_format. :return: The format string. :rtype: str """ return self._logger_format @logger_format.setter def logger_format(self, value): """The logger format. The logger_formatter will be updated when sets logger_format. :param value: The format string. :type value: str """ self._logger_format = value self.logger_formatter = logging.Formatter(self._logger_format) @property def retry_backoff_factor(self): """Retry backoff factor. :return: The backoff factor, float :rtype: float """ return self._retry_backoff_factor @retry_backoff_factor.setter def retry_backoff_factor(self, value): """Retry backoff factor. :param value: The backoff factor used to calculate intervals between retry attempts :type value: float """ if value < 2: raise ValueError("Retry backoff factor cannot be smaller than 2") self._retry_backoff_factor = value def get_api_key_with_prefix(self, identifier, alias=None): """Gets API key (with prefix if set). :param identifier: The identifier of apiKey. :param alias: The alternative identifier of apiKey. :return: The token for api key authentication. """ if self.refresh_api_key_hook is not None: self.refresh_api_key_hook(self) key = self.api_key.get(identifier, self.api_key.get(alias) if alias is not None else None) if key: prefix = self.api_key_prefix.get(identifier) if prefix: return "%s %s" % (prefix, key) return key def get_basic_auth_token(self): """Gets HTTP basic authentication header (string). :return: The token for basic HTTP authentication. """ username = "" if self.username is not None: username = self.username password = "" if self.password is not None: password = self.password return urllib3.util.make_headers(basic_auth=username + ":" + password).get("authorization") def get_host_settings(self): """Gets an array of host settings :return: An array of host settings """ return [ { "url": "https://{subdomain}.{site}", "description": "No description provided", "variables": { "site": { "description": "The regional site for Datadog customers.", "default_value": "datadoghq.com", "enum_values": [ "datadoghq.com", "us3.datadoghq.com", "us5.datadoghq.com", "ap1.datadoghq.com", "datadoghq.eu", "ddog-gov.com", ], }, "subdomain": { "description": "The subdomain where the API is deployed.", "default_value": "api", }, }, }, { "url": "{protocol}://{name}", "description": "No description provided", "variables": { "name": { "description": "Full site DNS name.", "default_value": "api.datadoghq.com", }, "protocol": { "description": "The protocol for accessing the API.", "default_value": "https", }, }, }, { "url": "https://{subdomain}.{site}", "description": "No description provided", "variables": { "site": { "description": "Any Datadog deployment.", "default_value": "datadoghq.com", }, "subdomain": { "description": "The subdomain where the API is deployed.", "default_value": "api", }, }, }, ] def get_host_from_settings(self, index, variables=None, servers=None): """Gets host URL based on the index and variables. :param index: Array index of the host settings. :param variables: Hash of variable and the corresponding value. :param servers: An array of host settings or None. :return: URL based on host settings. """ if index is None: return self._base_path variables = {} if variables is None else variables servers = self.get_host_settings() if servers is None else servers try: server = servers[index] except IndexError: raise ValueError( "Invalid index {} when selecting the host settings. " "Must be less than {}".format(index, len(servers)) ) url = server["url"] # go through variables and replace placeholders for variable_name, variable in server.get("variables", {}).items(): used_value = variables.get(variable_name, variable["default_value"]) if "enum_values" in variable and used_value not in variable["enum_values"]: raise ValueError( "The variable `{}` in the host URL has invalid value " "{}. Must be {}.".format( variable_name, variables[variable_name], variable["enum_values"] ) ) url = url.replace(f"{{{variable_name}}}", used_value) return url @property def host(self): """Return generated host.""" return self.get_host_from_settings(self.server_index, variables=self.server_variables) @host.setter def host(self, value): """Fix base path.""" self._base_path = value self.server_index = None def auth_settings(self): """Gets Auth Settings dict for api client. :return: The Auth Settings information dict. """ auth = {} if self.access_token is not None: auth["AuthZ"] = { "type": "oauth2", "in": "header", "key": "Authorization", "value": "Bearer " + self.access_token, } if "apiKeyAuth" in self.api_key: auth["apiKeyAuth"] = { "type": "api_key", "in": "header", "key": "DD-API-KEY", "value": self.get_api_key_with_prefix( "apiKeyAuth", ), } if "appKeyAuth" in self.api_key: auth["appKeyAuth"] = { "type": "api_key", "in": "header", "key": "DD-APPLICATION-KEY", "value": self.get_api_key_with_prefix( "appKeyAuth", ), } return auth