Source code for

# coding: utf-8

# Copyright (C) 2018 Orange
# This software is distributed under the terms and conditions of the 'Apache-2.0'
# license which can be found in the '' file
# or at ''.

"""Functions and classes used by the main categories of methods."""

from __future__ import unicode_literals
import json
import sys

import requests
from requests.exceptions import HTTPError

[docs]class RtpyBase(object): """ Rtpy Base class. Attributes ---------- _prefix: str API endpoint used for a given API method category ("search/"...) _category: str Major API method category (Searches, Repositories...) _user_settings: dict Complete user settings, default values are changed at class instantiation """ def __init__(self, provided_settings, prefix, category): """ Object instantiation. User settings get set and verified, _category and _prefix attributes are set. Parameters ---------- provided_settings: str Original settings dict supplied by the user from the rtpy.Rtpy class prefix: str API endpoint used for a given API method category ("search/"...) category: str Major API method category (Searches, Repositories...) """ self._category = category self._prefix = prefix self._user_settings = { "af_url": None, "api_key": None, "username": None, "password": None, "raw_response": False, "verbose_level": 0, "auth": (), "X-JFrog-Art-Api": None, "api_endpoint": None, "session": requests.Session(), } self._configure_user_settings(provided_settings) self._validate_user_settings() def _check_provided_keys_in_settings(self, provided_settings): """ Check if the provided settings only modify the correct fields. Parameters ---------- provided_settings: str Original settings dict supplied by the user from the rtpy.Rtpy class Raises ------ UserSettingsError If the settings are invalid Returns ------- None Nothing """ allowed_settings_keys = [ "af_url", "api_key", "username", "password", "raw_response", "verbose_level", "session", ] message = "" for setting in allowed_settings_keys: message += '"' + str(setting) + '" ' message += ( "are the only settings " "that can provided!\n" "Offending key(s) supplied : " ) offending_keys = ( key for key in provided_settings if key not in allowed_settings_keys ) offending_keys = tuple(key for key in offending_keys) for key in offending_keys: message += '"{}" ' message = message.format(*offending_keys) if offending_keys: raise UserSettingsError(message) def _configure_user_settings(self, provided_settings): """ Configure or update user settings. A _modified_user_settings dict attribute is used to store the last modified settings For each setting: If the value is different from default then update the _user_settings dict If the value is different from default then add the setting to the _modified_user_settings attribute Returns ------- None Nothing """ self._check_provided_keys_in_settings(provided_settings) self._modified_user_settings = {} for setting in provided_settings: if setting != self._user_settings[setting]: self._user_settings[setting] = provided_settings[setting] self._modified_user_settings[setting] = self._user_settings[setting] def _validate_user_settings(self): """ Verify if the _user_settings attribute dict is valid. Raises ------ UserSettingsError If the _user_settings dict is invalid Returns ------- None Nothing """ if ( not self._user_settings["af_url"] or not self._user_settings["api_key"] and ( not self._user_settings["username"] or not self._user_settings["password"] ) ): message = ( "An af_url, and api_key or username and password " + "must be provided in a dictionary to set the user settings!" ) raise UserSettingsError(message) if self._user_settings["api_key"] and ( self._user_settings["username"] or self._user_settings["password"] ): message = ( "An api_key and user name and password" + " can't be provided at the same time!" ) raise UserSettingsError(message) self._user_settings["auth"] = None if self._user_settings["username"] and self._user_settings["password"]: self._user_settings["auth"] = ( self._user_settings["username"], self._user_settings["password"], ) self._user_settings["X-JFrog-Art-Api"] = self._user_settings["api_key"] self._user_settings["api_endpoint"] = self._user_settings["af_url"] + "/api/" allowed_raw_response = [False, True] allowed_verbose_level = [0, 1] if self._user_settings["raw_response"] not in allowed_raw_response: raise UserSettingsError( "raw_response must be " + str(allowed_raw_response) + "!" ) if self._user_settings["verbose_level"] not in allowed_verbose_level: raise UserSettingsError( "verbose_level must be " + str(allowed_verbose_level) + "!" ) def _request( self, verb, target, api_method, kwargs, byte_output=False, no_api=False, data=None, params=None, ): """ Call the remote API, process the response and return it. Parameters ---------- verb: str HTTP verb ("GET", "POST"...) target: str API sub endpoint specific for the method api_method: str Name of the specific method (category and name) kwargs: dict Dictionary of keyword arguments (supplied by the method) byte_output: bool True will return a raw requests.Response() object True indicates that the result of the method will be used to download a file/artifact False by default no_api: bool True to remove 'api/' from the target endpoint False by default data Python requests data keyword argument params Python requests json keyword argument Returns ------- response The result of the processed request.Response() given by _convert_response method """ # Checking if the settings kwarg was supplied settings = self._settings_if_settings_in_kwargs(kwargs) if settings: self._configure_user_settings(settings) self._validate_user_settings() headers = {} if self._user_settings["X-JFrog-Art-Api"]: headers = {"X-JFrog-Art-Api": self._user_settings["X-JFrog-Art-Api"]} auth = self._user_settings["auth"] raw_response = self._user_settings["raw_response"] # Changing the endpoint when necessary if not no_api: request_url = self._user_settings["api_endpoint"] + target if no_api: request_url = self._user_settings["af_url"] + target possible_headers = [ "Content-Type", "X-Checksum-Deploy", "X-Checksum-Sha1", "X-Checksum-Sha256", "X-Result-detail", "X-GPG-PASSPHRASE", ] # Extracting the headers from the params dict if params: headers_to_add = [field for field in params if field in possible_headers] for header in headers_to_add: headers[header] = str(params[header]) del params[header] else: params = None if self._user_settings["verbose_level"] >= 1: message = ( "\n\nPerforming Artifactory REST API operation : " + api_method + "\nVerb : " + verb + "\nURL : " + request_url ) sys.stdout.write(message) response = self._user_settings["session"].request( verb, request_url, headers=headers, json=params, data=data, auth=auth ) if self._user_settings["verbose_level"] >= 1: message = "\nStatus Code : " + str(response.status_code) + "\n" sys.stdout.write(message) if settings: self._configure_user_settings(self._modified_user_settings) self._validate_user_settings() return self._convert_response( api_method, request_url, verb, response, raw_response, byte_output ) def _convert_response( self, api_method, target, verb, response, raw_response, byte_output ): """ Convert (if necessary) a requests.Response() object raise an error. If response.text is of json format, a Python dict will be returned or an error if that dict is an error output dict Else a raw response.text will be returned A raw Python Requests response object is returned if raw_response or byte_output is true Parameters ---------- api_method: str Name of the specific method (category and name) target: str API sub endpoint specific for the method verb: str HTTP verb ("GET", "POST"...) response: requests.Response The requests.Response object for the API call raw_response: bool True to return the original requests.Response byte_output: bool True to return the original requests.Response if no errors are found Raises ------ error An error raised by the _process_and_raise_error method Returns ------- response: requests.Response or str or dict The returned response to the client """ if raw_response: return response # Trying to see if the response content can be loaded as a json bad_status_codes = list(range(400, 600)) if response.status_code in bad_status_codes: self._process_and_raise_error(response, api_method, target, verb) if not raw_response: try: if byte_output: return response else: response2 = response.json() return response2 except ValueError: return response.text def _process_and_raise_error(self, response, api_method, target, verb): """ Look for errors in a requests.Response object from the Artifactory REST API. Parameters ---------- response: requests.Response() Original response for the API call api_method: str Name of the specific method (category and name) target: str API sub endpoint specific for the method verb: str HTTP verb ("GET", "POST"...) Raises ------ MalformedAfApiError If there is an error but the JSON is malformed AfApiError If the is an error and the JSON is a standard Artifactory REST API error JSON """ malformed_json = False error_details = None try: json_content = response.json() if "errors" in json_content: message = json_content["errors"][0]["message"] if "status" in json_content["errors"][0]: status_code = json_content["errors"][0]["status"] else: status_code = response.status_code error_details = { "api_method": api_method, "url": target, "verb": verb, "status_code": status_code, "message": message, } if not error_details: malformed_json = True except ValueError: malformed_json = True if malformed_json: message = ( "The json output for the error was malformed " "(Not a standard Artifactory REST API error json), " 'set the "raw_response" setting to True in your ' "settings dictionary to get a raw requests.Response() object " "for debugging!, requests.Response.text : {}".format(response.text) ) raise self.MalformedAfApiError(message) else: raise self.AfApiError(error_details) def _append_to_string(self, target, options): """ Add a string of options to a target string (URL in this case). Parameters ---------- target: str API sub endpoint specific for the method options: str String of options Returns ------- target: str The target with the string of options appended to it """ if options: target += options return target return target def _add_forward_slash_if_not_empty(self, target, path): """ Add a forward slash to a string if the other string != "". Parameters ---------- target: str API sub endpoint specific for the method path: str Path in the sub endpoint (usually and item) """ if path != "": target += "/" + path return target return target def _settings_if_settings_in_kwargs(self, kwargs): """ Extract the settings key from the kwargs dict and return it if present. Parameters ---------- kwargs: dict Keyword arguments Returns ------- settings: dict or None The settings dict if in kwargs, None otherwise """ if "settings" in kwargs: settings = kwargs["settings"] else: settings = None return settings
[docs] class MalformedAfApiError(HTTPError): """Raised when encountering a malformed Artifactory REST API error JSON.""" pass
[docs] class AfApiError(HTTPError): """ Raised when encountering a standard Artifactory REST API error JSON. Attributes ---------- api_method: str Name of the specific method (category and name) url: str Full URL used for API call verb: str HTTP verb used for the API call ("GET", "POST"...) status_code: int HTTP status code for the API call message: str Error message given by the Artifactory REST API print_message: str Well formatted multiline message with all the attributes """ def __init__(self, error_details): """Error object instantiation.""" self.api_method = error_details["api_method"] self.url = error_details["url"] self.verb = error_details["verb"] self.status_code = error_details["status_code"] self.message = error_details["message"] self.print_message = ( "\nArtifactory REST API operation : " + self.api_method + "\nURL : " + self.url + "\nVerb : " + self.verb + "\nStatus Code : " + str(self.status_code) + "\nMessage : " + self.message ) HTTPError.__init__(self, self.print_message)
[docs] class RtpyError(ValueError): """ Raised if arguments are wrong or incomplete for a method. This error is used as a preemptive measure in some methods to check arguments. """ pass
[docs]class UserSettingsError(ValueError): """Raised if some of the provided settings are incorrect or missing.""" pass
[docs]def json_to_dict(json_file_path): """ Convert a .json file to a Python dictionary. Parameters ---------- json_file_path: str Path of the JSON file Returns ------- dictionary: dict The original JSON file as a Python dictionary """ with open(json_file_path, "r") as json_data: dictionary = json.load(json_data, encoding="utf8") return dictionary