PK!GArtpy/__init__.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Exposed objects/functions to end user.""" from .rtpy import Rtpy from .tools import json_to_dict, UserSettingsError __all__ = ["Rtpy", "json_to_dict", "UserSettingsError"] PK!CtVtVrtpy/artifacts_and_storage.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the ARTIFACTS AND STORAGE REST API Methods category.""" from .tools import RtpyBase class RtpyArtifactsAndStorage(RtpyBase): """ARTIFACTS AND STORAGE methods category.""" def folder_info(self, repo_key, folder_path, **kwargs): """ Folder Info. For virtual use, the virtual repository returns the unified children. Supported by local, local-cached and virtual repositories. Parameters ---------- repo_key: str Key of the repository folder_path: str The path of the folder in the repository **kwargs Keyword arguments """ api_method = self._category + "Folder Info" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, folder_path) return self._request("GET", target, api_method, kwargs) def file_info(self, repo_key, file_path, **kwargs): """ File Info. For virtual use the virtual repository returns the resolved file. Supported by local, local-cached and virtual repositories. Parameters ---------- repo_key: str Key of the repository file_path: str The path of the file in the repository **kwargs Keyword arguments """ api_method = self._category + "File Info" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, file_path) return self._request("GET", target, api_method, kwargs) def get_storage_summary_info(self, **kwargs): """ Return storage summary information. regarding binaries file store and repositories. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Get Storage Summary Info" target = "storageinfo" return self._request("GET", target, api_method, kwargs) def item_last_modified(self, repo_key, item_path, **kwargs): """ Retrieve the last modified item at the given path. If the given path is a folder, the latest last modified item is searched for recursively. Supported by local and local-cached repositories. Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository **kwargs Keyword arguments """ api_method = self._category + "Item Last Modified" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, item_path) target = target + "?lastModified" return self._request("GET", target, api_method, kwargs) def file_statistics(self, repo_key, item_path, **kwargs): """ Item statistics. Record the number of times an item was downloaded, last download date and last downloader. Supported by local and local-cached repositories. Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository **kwargs Keyword arguments """ api_method = self._category + "File Statistics" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, item_path) target = target + "?stats" return self._request("GET", target, api_method, kwargs) def item_properties(self, repo_key, item_path, properties=None, **kwargs): """ Item Properties. Optionally return only the properties requested. Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository properties: str String of properties **kwargs Keyword arguments """ api_method = self._category + "Item Properties" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, item_path) target = target + "?properties" if properties: target = target + "=" + properties return self._request("GET", target, api_method, kwargs) def set_item_properties( self, repo_key, item_path, properties, options=None, **kwargs ): """ Attach properties to an item (file or folder). When a folder is used property attachment is recursive by default. Supported by local and local-cached repositories Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository properties: str String of properties options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "Set Item Properties" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, item_path) target = target + "?properties=" + properties target = self._append_to_string(target, options) return self._request("PUT", target, api_method, kwargs) def delete_item_properties( self, repo_key, item_path, properties, options=None, **kwargs ): """ Delete the specified properties from an item (file or folder). When a folder is used property removal is recursive by default. Supported by local and local-cached repositories. Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository properties: str String of properties options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "Delete Item Properties" target = self._prefix + repo_key target = self._add_forward_slash_if_not_empty(target, item_path) target = target + "?properties=" + properties target = self._append_to_string(target, options) return self._request("DELETE", target, api_method, kwargs) def set_item_sha256_checksum(self, params, **kwargs): """ Calculate an artifact's SHA256 checksum and attaches it as a property. (with key "sha256"). If the artifact is a folder, then recursively calculates the SHA256 of each item in the folder and attaches the property to each item. Parameters ---------- params: dict Dictionary comprised of {"repo_key": str, "path": str} **kwargs Keyword arguments """ api_method = self._category + "Set Item SHA256 Checksum" target = "checksum/sha256" params["Content-Type"] = "application/json" return self._request("POST", target, api_method, kwargs, params=params) def retrieve_artifact(self, repo_key, artifact_path, **kwargs): """ Retrieve an artifact from the specified destination. Parameters ---------- repo_key: str Key of the repository artifact_path: str Path of the artifact in the repository **kwargs Keyword arguments """ api_method = self._category + "Retrieve Artifact" if artifact_path == "": message = "artifact path can't be empty !" raise self.RtpyError(message) target = "/" + repo_key + "/" + artifact_path return self._request( "GET", target, api_method, kwargs, byte_output=True, no_api=True ) # Unsupported methods # def retrieve_latest_artifact(): # def retrieve_build_artifacts_archive(): def retrieve_folder_or_repository_archive( self, repo_key, path, archive_type, include_checksums=False, **kwargs ): """ Retrieve an archive file (supports zip/tar/tar.gz/tgz). containing all the artifacts that reside under the specified path (folder or repository root). Requires Enable Folder Download to be set. Parameters ---------- repo_key: str Key of the repository path: str Path of the folder in the repository archive_type: str Type of archive include_checksums: bool, optional True to include checksums, False by default **kwargs Keyword arguments """ api_method = self._category + "Retrieve Folder or Repository Archive" if archive_type not in ["zip", "tar", "tar.gz", "tgz"]: message = "archive_type must be zip, tar, tar.gz or tgz !" raise self.RtpyError(message) target = "archive/download/" + repo_key target = self._add_forward_slash_if_not_empty(target, path) target = target + "?archiveType=" + archive_type if include_checksums: target = target + "&includeChecksumFiles=true" if not include_checksums: target = target + "&includeChecksumFiles=false" return self._request("GET", target, api_method, kwargs, byte_output=True) def trace_artifact_retrieval(self, repo_key, item_path, **kwargs): """ Simulate an artifact retrieval request from the specified. location and returns verbose output about the resolution process. Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository **kwargs Keyword arguments """ api_method = self._category + "Trace Artifact Retrieval" target = "/" + repo_key + "/" + item_path + "?trace" return self._request("GET", target, api_method, kwargs, no_api=True) # def archive_entry_download(): def create_directory(self, repo_key, directory_path, **kwargs): """ Create new directory at the specified destination. Parameters ---------- repo_key: str Key of the repository directory_path: str Path of the directory in the repository **kwargs Keyword arguments """ api_method = self._category + "Create Directory" target = "/" + repo_key + "/" + directory_path + "/" return self._request("PUT", target, api_method, kwargs, no_api=True) def deploy_artifact( self, repo_key, local_artifact_path, target_artifact_path, **kwargs ): """ Deploy an artifact to the specified destination. Parameters ---------- repo_key: str Key of the repository local_artifact_path: str Local path of the artifact to upload target_artifact_path: str Target path of the artifact in the repository **kwargs Keyword arguments """ api_method = self._category + "Deploy Artifact" target = "/" + repo_key + "/" + target_artifact_path with open(local_artifact_path, "rb") as files: return self._request( "PUT", target, api_method, kwargs, data=files, no_api=True ) def deploy_artifact_by_checksum( self, repo_key, target_artifact_path, sha_type, sha_value, **kwargs ): """ Deploy an artifact to the specified destination. By checking if the artifact content already exists in Artifactory. If Artifactory already contains a user readable artifact with the same checksum the artifact content is copied over to the new location and return a response without requiring content transfer. Otherwise, a 404 error is returned to indicate that content upload is expected in order to deploy the artifact. Parameters ---------- repo_key: str Key of the repository target_artifact_path: str Target path of the artifact in the repository sha_type: str Type of secure hash sha_value: str Value of the secure hash **kwargs Keyword arguments """ api_method = self._category + "Deploy Artifact By Checksum" if sha_type not in ["sha1", "sha256"]: message = ( 'sha_type must be "sha1" or "sha256", ' + 'type given was "' + sha_type + '"' ) raise self.RtpyError(message) params = {} params["X-Checksum-Deploy"] = True if sha_type == "sha1": params["X-Checksum-Sha1"] = sha_value if sha_type == "sha256": params["X-Checksum-Sha256"] = sha_value target = "/" + repo_key + "/" + target_artifact_path return self._request( "PUT", target, api_method, kwargs, no_api=True, params=params ) # Unsupported methods # def deploy_artifacts_from_archive() # def push_a_set_of_artifacts_to_bintray() # def push_docker_tag_to_bintray() # def distribute_artifact() # def file_compliance_info() def delete_item(self, repo_key, path_to_item, **kwargs): """ Delete a file or a folder from the specified destination. Parameters ---------- repo_key: str Key of the repository path_to_item: str Path of the item in the repository **kwargs Keyword arguments """ api_method = self._category + "Delete Item" target = "/" + repo_key + "/" + path_to_item return self._request("DELETE", target, api_method, kwargs, no_api=True) def copy_item( self, src_repo_key, src_item_path, target_repo_key, target_item_path, options=None, **kwargs ): """ Copy an artifact or a folder to the specified destination. Supported by local repositories only. Parameters ---------- src_repo_key: str Key of the source repository src_item_path: str Path of the item in the source repository target_repo_key: str Key of the target repository target_item_path: str Path of the item in the target repository options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "Copy Item" target = "copy/" + src_repo_key target = self._add_forward_slash_if_not_empty(target, src_item_path) target = target + "?to=/" + target_repo_key if target_repo_key != "": target = target + "/" + target_item_path target = self._append_to_string(target, options) return self._request("POST", target, api_method, kwargs) def move_item( self, src_repo_key, src_item_path, target_repo_key, target_item_path, options=None, **kwargs ): """ Move an artifact or a folder to the specified destination. Supported by local repositories only. Parameters ---------- src_repo_key: str Key of the source repository src_item_path: str Path of the item in the source repository target_repo_key: str Key of the target repository target_item_path: str Path of the item in the target repository options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "Move Item" target = "move/" + src_repo_key target = self._add_forward_slash_if_not_empty(target, src_item_path) target = target + "?to=/" + target_repo_key if target_repo_key != "": target = target + "/" + target_item_path target = self._append_to_string(target, options) return self._request("POST", target, api_method, kwargs) # Unsupported methods # def get_repository_replication_configuration(): # def set_repository_replication_configuration(): # def update_repository_replication_configuration(): # def delete_repository_replication_configuration(): # def scheduled_replication_status(): # def pull_or_push_replication(): # def update_local_multi_push_replication(): # def delete_local_multi_push_replication(): # def enable_or_disable_multiple_replications(): # def global_system_replication_configuration(): # def block_system_replication(): # def unblock_system_replication(): def artifact_sync_download(self, repo_key, artifact_path, options=None, **kwargs): """ Download an artifact. With or without returning he actual content to the client. When tracking the progress marks are printed (by default every 1024 bytes). This is extremely useful if you want to trigger downloads on a remote Artifactory server, for example to force eager cache population of large artifacts, but want to avoid the bandwidth consumption involved in transferring the artifacts to the triggering client. If no content parameter is specified the file content is downloaded to the client. Parameters ---------- repo_key: str Key of the repository artifact_path: str Path of the artifact in the repository options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "Artifact Sync Download" if artifact_path == "": message = "artifact path can't be empty !" raise self.RtpyError(message) target = "download/" + repo_key + "/" + artifact_path target = self._append_to_string(target, options) return self._request("GET", target, api_method, kwargs) def file_list(self, repo_key, folder_path, options=None, **kwargs): """ Get a flat (the default) or deep listing of the files and folders. (not included by default) within a folder For deep listing you can specify an optional depth to limit the results. Optionally include a map of metadata timestamp values as part of the result Parameters ---------- repo_key: str Key of the repository folder_path: str Path of the folder in the repository options: str, optional String of options **kwargs Keyword arguments """ api_method = self._category + "File List" target = "storage/" + repo_key target = self._add_forward_slash_if_not_empty(target, folder_path) target = target + "?list" target = self._append_to_string(target, options) return self._request("GET", target, api_method, kwargs) def get_background_tasks(self, **kwargs): """ Retrieve list of background tasks currently scheduled. Or running in Artifactory In HA, the nodeId is added to each task. Task can be in one of few states: scheduled, running, stopped, canceled. Running task also shows the task start time. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Get Background Tasks" target = "tasks" return self._request("GET", target, api_method, kwargs) def empty_trash_can(self, **kwargs): """ Empty the trash can permanently deleting all its current contents. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Empty Trash Can" target = "trash/empty" return self._request("POST", target, api_method, kwargs) def delete_item_from_trash_can(self, path_in_trashcan, **kwargs): """ Permanently delete an item from the trash can. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Delete Item From Trash Can" target = "trash/clean/" + path_in_trashcan return self._request("DELETE", target, api_method, kwargs) def restore_item_from_trash_can(self, path_in_trashcan, target_path, **kwargs): """ Restore an item from the trash can. Parameters ---------- path_in_trashcan: str Path of the item in the trashcan (repo_name/folder/file) target_path: str Where to restore the item (repo_name/folder/file) """ api_method = self._category + "Restore Item From Trash Can" target = "trash/restore/" + path_in_trashcan + "?to=" + target_path return self._request("POST", target, api_method, kwargs) def optimize_system_storage(self, **kwargs): """ Raise a flag to invoke balancing between redundant storage units. Of a sharded filestore following the next garbage collection. """ api_method = self._category + "Optimize System Storage" target = "system/storage/optimize" return self._request("POST", target, api_method, kwargs) # Unsupported methods # def get_puppet_modules() # def get_puppet_module() # def get_puppet_releases() # def get_puppet_release() PK!;Prtpy/builds.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the BUILDS REST API Methods category.""" from .tools import RtpyBase class RtpyBuilds(RtpyBase): """BUILDS methods category.""" def all_builds(self, **kwargs): """Provide information on all builds. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "All Builds" return self._request("GET", "build", api_method, kwargs) # Unsupported methods # def build_runs() # def build_upload() # def build_info() # def builds_diff() # def build_promotion() # def promote_docker_image() # def delete_builds() # def push_build_to_bintray() # def distribute_build() # def control_build_retention() PK!33rtpy/import_and_export.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the IMPORT AND EXPORT REST API Methods category.""" PK!P" ))rtpy/plugins.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the PLUGINS REST API Methods category.""" PK! U憞66rtpy/repositories.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the REPOSITORIES REST API Methods category.""" from .tools import RtpyBase class RtpyRepositories(RtpyBase): """REPOSITORIES methods category.""" def get_repositories(self, options=None, **kwargs): """ Return a list of minimal repository details. For all repositories of the specified type. Parameters ---------- options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Get Repositories" target = self._append_to_string("repositories", options) return self._request("GET", target, api_method, kwargs) def repository_configuration(self, repo_key, **kwargs): """ Retrieve the current configuration of a repository. Supported by local, remote and virtual repositories. Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Repository Configuration" target = self._prefix + repo_key return self._request("GET", target, api_method, kwargs) def create_repository(self, params, **kwargs): """ Create a new repository in Artifactory with the provided configuration. Supported by local, remote and virtual repositories. A position may be specified using the pos parameter. If the map size is shorter than pos the repository is the last one (the default behavior). Parameters ---------- params: dict Parameters of the repository **kwargs Keyword arguments """ api_method = self._category + "Create Repository" repo_key = params["key"] target = self._prefix + repo_key return self._request("PUT", target, api_method, kwargs, params=params) def update_repository_configuration(self, params, **kwargs): """ Update an exiting repository configuration in Artifactory. With the provided configuration elements. Supported by local, remote and virtual repositories. Parameters ---------- params: dict Parameters of the repository **kwargs Keyword arguments """ api_method = self._category + "Update Repository Configuration" repo_key = params["key"] target = self._prefix + repo_key params["Content-Type"] = "application/json" return self._request("POST", target, api_method, kwargs, params=params) def delete_repository(self, repo_key, **kwargs): """ Remove a repository. Configuration together with the whole repository content. Supported by local, remote and virtual repositories Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Delete Repository" target = self._prefix + repo_key return self._request("DELETE", target, api_method, kwargs) def calculate_yum_repository_metadata( self, repo_key, x_gpg_passphrase=None, options=None, **kwargs ): """ Calculate/recalculate the YUM metdata for a repository. Based on the RPM package currently hosted in the repository. Supported by local and virtual repositories only. Calculation can be synchronous (the default) or asynchronous. For Virtual repositories, calculates the merged metadata from all aggregated repositories on the specified path. The path parameter must be passed for virtual calculation. Parameters ---------- repo_key: str Key of the repository x_gpg_passphrase: str Passphrase options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate YUM Repository Metadata" target = "yum/" + repo_key target = self._append_to_string(target, options) params = {} if x_gpg_passphrase: params["X-GPG-PASSPHRASE"] = x_gpg_passphrase return self._request("POST", target, api_method, kwargs, params=params) def calculate_nuget_repository_metadata(self, repo_key, **kwargs): """ Recalculate all the NuGet packages for a repository. (local/cache/virtual), and re-annotate the NuGet properties for each NuGet package according to it's internal nuspec file. Please see the NuGet integration documentation for more details. Supported by local, local-cache, remote and virtual repositories. Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Calculate NuGet Repository Metadata" target = "nuget/" + repo_key + "/reindex" return self._request("POST", target, api_method, kwargs) def calculate_npm_repository_metadata(self, repo_key, **kwargs): """ Recalculate the npm search index for this repository (local/virtual). Please see the Npm integration documentation for more details. Supported by local and virtual repositories. Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Calculate Npm Repository Metadata" target = "npm/" + repo_key + "/reindex" return self._request("POST", target, api_method, kwargs) def calculate_maven_index(self, options, **kwargs): """ Calculates/caches a Maven index for the specified repositories. For a virtual repository specify all underlying repositories that you want the aggregated index to include. Calculation can be forced, which for remote repositories will cause downloading of a remote index even if a locally ached index has not yet expired; and index recalculation based on the cache on any failure to download the remote index, including communication errors (the default behavior is to only use the cache when a remote index cannot be found and returns a 404). Forcing has no effect on local repositories index calculation. Parameters ---------- options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate Maven Index" target = "maven?" + options return self._request("POST", target, api_method, kwargs) def calculate_maven_metadata(self, repo_key, folder_path, options=None, **kwargs): """ Calculate Maven metadata on the specified path. (local repositories only). Parameters ---------- repo_key: str Key of the repository folder_path: str Path of the folder in the repository options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate Maven Metadata" target = "maven/calculateMetadata/" + repo_key + "/" + folder_path target = self._append_to_string(target, options) return self._request("POST", target, api_method, kwargs) # Unsupported ressource intensive methods def calculate_debian_repository_metadata( self, repo_key, x_gpg_passphrase=None, options=None, **kwargs ): """ Calculate/recalculate the Packages and Release metadata. for this repository, based on the Debian packages in it. Calculation can be synchronous (the default) or asynchronous. Please refer to Debian Repositories for more details. Supported by local repositories only. From version 4.4, by default, the recalculation process also writes several entries from the Debian package's metadata as properties on all of the artifacts (based on the control file's content). This operation may not always be required (for example, if the Debian files are intact and were not modified, only the index needs to be recalculated. The operation is resource intensive and can be disabled by passing the ?writeProps=0 query param. From version 5.7, the target repository can be a virtual repository. Parameters ---------- repo_key: str Key of the repository x_gpg_passphrase: str Passphrase options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate Debian Repository Metadata" target = "deb/reindex/" + repo_key target = self._append_to_string(target, options) params = {} if x_gpg_passphrase: params["X-GPG-PASSPHRASE"] = x_gpg_passphrase return self._request("POST", target, api_method, kwargs, params=params) def calculate_opkg_repository_metadata( self, repo_key, x_gpg_passphrase=None, options=None, **kwargs ): """ Calculate/recalculate the Packages and Release metadata fora repository. Based on the ipk packages in it (in each feed location). Calculation can be synchronous (the default) or asynchronous. Please refer to Opkg Repositories for more details. Supported by local repositories only. By default, the recalculation process also writes several entries from the ipk package's metadata as properties on all of the artifacts (based on the control file's content). This operation may not always be required (for example, if the ipk files are intact and were not modified, only the index needs to be recalculated. The operation is resource intensive and can be disabled by passing the ?writeProps=0 query param. Parameters ---------- repo_key: str Key of the repository x_gpg_passphrase: str Passphrase options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate Opkg Repository Metadata" target = "opkg/reindex/" + repo_key target = self._append_to_string(target, options) params = {} if x_gpg_passphrase: params["X-GPG-PASSPHRASE"] = x_gpg_passphrase return self._request("POST", target, api_method, kwargs, params=params) def calculate_bower_index(self, repo_key, **kwargs): """ Recalculate the index for a Bower repository. Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Calculate Bower Index" target = "bower/" + repo_key + "/" + "reindex" return self._request("POST", target, api_method, kwargs) def calculate_helm_chart_index(self, repo_key, **kwargs): """ Calculate Helm chart index on the specified path. (local repositories only). Parameters ---------- repo_key: str Key of the repository **kwargs Keyword arguments """ api_method = self._category + "Calculate Helm Chart Index" target = "helm/" + repo_key + "/" + "reindex" return self._request("POST", target, api_method, kwargs) def calculate_cran_repository_metadata(self, repo_key, options=None, **kwargs): """ Calculates/recalculates the Packages and Release metadata for a repository. Based on the CRAN packages in it. The calculation can be synchronous (the default) or asynchronous. Please refer to CRAN Repositories for more details. Supported by local repositories only. From version 6.1, by default, the recalculation process also writes several entries from the CRAN package's metadata as properties on all of the artifacts (based on the control file's content). Parameters ---------- repo_key: str Key of the repository options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate CRAN Repository Metadata" target = "cran/reindex/" + repo_key target = self._append_to_string(target, options) return self._request("POST", target, api_method, kwargs) def calculate_conda_repository_metadata(self, repo_key, options=None, **kwargs): """ Calculate/recalculate the Conda packages and release metadata for a repository. The calculation can be synchronous (the default) or asynchronous. Please refer to Conda Repositories for more details. Supported for local repositories only Parameters ---------- repo_key: str Key of the repository options: str String of options **kwargs Keyword arguments """ api_method = self._category + "Calculate Conda Repository Metadata" target = "conda/reindex/" + repo_key target = self._append_to_string(target, options) return self._request("POST", target, api_method, kwargs) PK!  rtpy/rtpy.py# coding: utf-8 # Author: Guillaume Renault # Author email: grenault.ext@orange.com # 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Rtpy class definition with it's attributes which is exposed to the end user.""" from copy import deepcopy from .tools import RtpyBase from .artifacts_and_storage import RtpyArtifactsAndStorage from .builds import RtpyBuilds from .repositories import RtpyRepositories from .searches import RtpySearches from .security import RtpySecurity from .support import RtpySupport from .system_and_configuration import RtpySystemAndConfiguration class Rtpy(RtpyBase): """ Main parent class. Attributes are the classes are the methods categories. Parameters --------- settings: dict The user settings, mandaroty keys are "af_url" and "api_key" or "username" and "password". Attributes ---------- artifacts_and_storage: rtpy.artifacts_and_storage.RtpyArtifactsAndStorage Category for multiple API methods buils: rtpy.builds.RtpyBuilds Category for multiple API methods repositories: rtpy.repositories.RtpyRepositories Category for multiple API methods searches: rtpy.searches.RtpySearches Category for multiple API methods security: rtpy.security.RtpySecurity Category for multiple API methods support: rtpy.support.RtpySupport Category for multiple API methods system_and_configuration: rtpy.system_and_configuration.RtpySystemAndConfiguration Category for multiple API methods settings: dict Previously supplied settings at class instantiation """ def __init__(self, settings): """Object Instantiation.""" settings = deepcopy(settings) self.artifacts_and_storage = RtpyArtifactsAndStorage( settings, "storage/", "[ARTIFACTS & STORAGE] : " ) self.builds = RtpyBuilds(settings, "build/", "[BUILDS] : ") self.repositories = RtpyRepositories( settings, "repositories/", "[REPOSITORIES] : " ) self.searches = RtpySearches(settings, "search/", "[SEARCHES] : ") self.security = RtpySecurity(settings, "security/", "[SECURITY] : ") self.support = RtpySupport(settings, "support/", "[SUPPORT] : ") self.system_and_configuration = RtpySystemAndConfiguration( settings, "system/", "[SYSTEM & CONFIGURATION] : " ) self.settings = settings def __call__(self): """Shortcut to all the system health ping method to verify connectivity.""" return self.system_and_configuration.system_health_ping() PK!=rtpy/searches.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the SEARCHES REST API Methods category.""" from .tools import RtpyBase class RtpySearches(RtpyBase): """SEARCHES methods category.""" def artifactory_query_language(self, query, **kwargs): """ Search items using the Artifactory Query Language (AQL). Parameters --------- query: str The AQL string **kwargs Keyword arguments """ api_method = self._category + "Artifactory Query Language" target = self._prefix + "aql" params = {"Content-Type": "text/plain"} return self._request( "POST", target, api_method, kwargs, data=query, params=params ) """Old previously supported methods (code not updated for current rtpy) def artifact_search_quick_search( artifact_name, options=None, x_result_detail=None, **kwargs): # Artifact search by part of file name. # Searches return file info URIs. # Can limit search to specific repositories (local or caches). api_method = self._category + "Artifact Search Quick Search" target = self._prefix + 'artifact?name=' + artifact_name params = {} if x_result_detail: if x_result_detail not in ['info', 'properties', 'info, properties']: message = "x_result_detail must be 'info', 'properties', " + \ "'info, properties'!" raise self.RtpyError(message) params['X-Result-detail'] = x_result_detail if options: target = target + options return request('GET', target, api_method, kwargs, params=params) # def archive_entries_search_class_search(): # def gavc_search(): def property_search(properties, options=None, x_result_detail=None, **kwargs): api_method = self._category + "Property Search" target = self._prefix + 'prop?' + properties params = {} if x_result_detail: if x_result_detail not in ['info', 'properties', 'info, properties']: message = "x_result_detail must be 'info', 'properties', " + \ "'info, properties'!" raise self.RtpyError(message) params['X-Result-detail'] = x_result_detail if options: target = target + options return request('GET', target, api_method, kwargs, params=params) def checksum_search(checksum_type, checksum_value, options=None, x_result_detail=None, **kwargs): api_method = self._category + "Checksum Search" if checksum_type not in ['md5', 'sha1', 'sha256']: message = 'sha_type must be "md5", "sha1" or "sha256", ' + \ 'type given was "'+checksum_type+'"' raise self.RtpyError(message) target = self._prefix + 'checksum?' + checksum_type + \ '=' + checksum_value params = {} if x_result_detail: if x_result_detail not in ['info', 'properties', 'info, properties']: message = "x_result_detail must be 'info', 'properties', " + \ "'info, properties'!" raise self.RtpyError(message) params['X-Result-detail'] = x_result_detail if options: target = target + options return request('GET', target, api_method, kwargs, params=params) def bad_checksum_search(checksum_type, options=None, **kwargs): api_method = self._category + "Bad Checksum Search" target = self._prefix + 'badChecksum?type=' + checksum_type if checksum_type not in ['md5', 'sha1']: message = 'sha_type must be "md5"or "sha1", ' + \ 'type given was "'+checksum_type+'"' raise self.RtpyError(message) if options: target = target + options return request('GET', target, api_method, kwargs) def artifacts_not_downloaded_since(not_used_since, options=None, **kwargs): api_method = self._category + "Artifacts Not Downloaded Since" target = self._prefix + 'usage?notUsedSince=' + not_used_since if options: target = target + options return request('GET', target, api_method, kwargs) """ # Unsupported methods # def artifacts_with_date_in_date_range() # def artifacts_created_in_date_range() # def pattern_search() # def builds_for_dependcy() # def license_search() # def artifact_version_search() # def artifact_latest_version_search_based_on_layout() # def artifact_latest_version_search_based_on_properties() # def build_artifacts_search() def list_docker_repositories(self, repo_key, options=None, **kwargs): """ List all Docker repositories (the registry's _catalog). (Hosted in an Artifactory Docker repository). Parameters ---------- repo_key: str Key of the repository options: str String of options **kwargs Keyword arguments """ api_method = self._category + "List Docker Repositories" target = "docker/" + repo_key + "/v2/_catalog" target = self._append_to_string(target, options) return self._request("GET", target, api_method, kwargs) def list_docker_tags(self, repo_key, image_path, options=None, **kwargs): """ List all tags of the specified Artifactory Docker repository. Parameters ---------- repo_key: str Key of the repository image_path: str Path of the image in the repository options: str String of options **kwargs Keyword arguments """ api_method = self._category + "List Docker Tags" target = "docker/" + repo_key + "/v2/" + image_path + "/tags/list" target = self._append_to_string(target, options) return self._request("GET", target, api_method, kwargs) PK!ݓH3H3rtpy/security.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the SECURITY REST API Methods category.""" from .tools import RtpyBase class RtpySecurity(RtpyBase): """SECURITY methods category.""" def get_users(self, **kwargs): """ Get the users list. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Get Users" return self._request("GET", "security/users", api_method, kwargs) def get_user_details(self, username, **kwargs): """ Get the details of an Artifactory user. Parameters --------- username: str Name of the user **kwargs Keyword arguments """ api_method = self._category + "Get User Details" target = self._prefix + "users/" + username return self._request("GET", target, api_method, kwargs) def get_user_encrypted_password(self, **kwargs): """ Get the encrypted password of the authenticated requestor. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Get User Encrypted Password" target = self._prefix + "encryptedPassword" return self._request("GET", target, api_method, kwargs) def create_or_replace_user(self, params, **kwargs): """ Create a new user in Artifactory or replaces an existing user. Parameters --------- params: dict Settings of the user **kwargs Keyword arguments """ api_method = self._category + "Create or Replace User" username = params["name"] target = self._prefix + "users/" + username params["Content-Type"] = "application/json" return self._request("PUT", target, api_method, kwargs, params=params) def update_user(self, params, **kwargs): """ Update an exiting user in Artifactory with the provided user details. Parameters --------- params: dict Settings of the user **kwargs Keyword arguments """ api_method = self._category + "Update User" username = params["name"] target = self._prefix + "users/" + username params["Content-Type"] = "application/json" return self._request("POST", target, api_method, kwargs, params=params) def delete_user(self, username, **kwargs): """ Remove an Artifactory user. Parameters --------- username: str Name of the user **kwargs Keyword arguments """ api_method = self._category + "Delete User" target = self._prefix + "users/" + username return self._request("DELETE", target, api_method, kwargs) # Unsupported Methods # def expire_password_for_a_single_user() # def expire_password_for_multiple_users() # def expire_password_for_all_users() # def unexpire_password_for_a_single_user() # def change_password() # def get_password_expiration_policy() # def set_password_expiration_policy() # def configure_user_lock_policy() # def retrieve_user_lock_policy() def get_locked_out_users(self, **kwargs): """ Get a list of the locked out users. If locking out users is enabled, lists all users that were locked out due to recurrent incorrect login attempts. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Get Locked Out Users" target = self._prefix + "lockedUsers" return self._request("GET", target, api_method, kwargs) def unlock_locked_out_user(self, username, **kwargs): """ Unlock a single user that was locked out. Parameters --------- username: str Name of the user **kwargs Keyword arguments """ api_method = self._category + "Unlock Locked Out User" target = self._prefix + "unlockUsers/" + username return self._request("POST", target, api_method, kwargs) def unlock_locked_out_users(self, user_list, **kwargs): """ Unlock a list of users that were locked out. Parameters --------- user_list: list List of str representing usernames **kwargs Keyword arguments """ api_method = self._category + "Unlock Locked Out Users" target = self._prefix + "unlockUsers" counter = 0 data = "[ " for user in user_list: if counter == len(user_list) - 1: data = data + '"' + user + '"' else: data = data + '"' + user + '", ' counter = counter + 1 data = data + " ]" params = {} params["Content-Type"] = "application/json" return self._request( "POST", target, api_method, kwargs, data=data, params=params ) def unlock_all_locked_out_users(self, **kwargs): """ Unlock all users that were locked out. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Unlock All Locked Out Users" target = self._prefix + "unlockAllUsers" return self._request("POST", target, api_method, kwargs) def create_api_key(self, **kwargs): """ Create an API key for the current user. Returns an error if API key already exists, use regenerate API key instead. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Create API key" target = self._prefix + "apiKey" return self._request("POST", target, api_method, kwargs) def regenerate_api_key(self, **kwargs): """ Regenerate an API key for the current user. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Regenerate API key" target = self._prefix + "apiKey" return self._request("PUT", target, api_method, kwargs) def get_api_key(self, **kwargs): """ Get the current user's own API key. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Get API key" target = self._prefix + "apiKey" return self._request("GET", target, api_method, kwargs) def revoke_api_key(self, **kwargs): """ Revoke the current user's API key. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Revoke API key" target = self._prefix + "apiKey" return self._request("DELETE", target, api_method, kwargs) def revoke_user_api_key(self, username, **kwargs): """ Revoke the API key of another user. Parameters --------- username: str Name of the user **kwargs Keyword arguments """ api_method = self._category + "Revoke User API key" target = self._prefix + "apiKey/" + username return self._request("DELETE", target, api_method, kwargs) def get_groups(self, **kwargs): """ Get the groups list. Parameters --------- **kwargs Keyword arguments """ api_method = self._category + "Get Groups" target = self._prefix + "groups" return self._request("GET", target, api_method, kwargs) def get_group_details(self, group_name, **kwargs): """ Get the details of an Artifactory Group. Parameters --------- group_name: str Name of the group **kwargs Keyword arguments """ api_method = self._category + "Get Group Details" target = self._prefix + "groups" + "/" + group_name return self._request("GET", target, api_method, kwargs) def create_or_replace_group(self, params, **kwargs): """ Create a new group in Artifactory or replace an existing group. Parameters --------- params: dict Settings of the group **kwargs Keyword arguments """ api_method = self._category + "Create or Replace Group" group_name = params["group_name"] target = self._prefix + "groups" + "/" + group_name return self._request("PUT", target, api_method, kwargs, params=params) def update_group(self, params, **kwargs): """ Update an exiting group in Artifactory. Parameters --------- params: dict Settings of the group **kwargs Keyword arguments """ api_method = self._category + "Update Group" group_name = params["group_name"] target = self._prefix + "groups" + "/" + group_name return self._request("POST", target, api_method, kwargs, params=params) def delete_group(self, group_name, **kwargs): """ Remove an Artifactory group. Parameters --------- group_name: str Name of the group **kwargs Keyword arguments """ api_method = self._category + "Delete Group" target = self._prefix + "groups" + "/" + group_name return self._request("DELETE", target, api_method, kwargs) def get_permission_targets(self, **kwargs): """ Get the permission targets list. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Get Permission Targets" target = self._prefix + "permissions" return self._request("GET", target, api_method, kwargs) def get_permission_target_details(self, permission_target_name, **kwargs): """ Get the details of an Artifactory Permission Target. Parameters ---------- permission_target_name: str Name of the permission target **kwargs Keyword arguments """ api_method = self._category + "Get Permission Target Details" target = self._prefix + "permissions/" + permission_target_name return self._request("GET", target, api_method, kwargs) def create_or_replace_permission_target(self, params, **kwargs): """ Create a new permission target in Artifactory. (or replace an existing permission target). Parameters ---------- params: dict Settings of the permission target **kwargs Keyword arguments """ api_method = self._category + "Create or Replace Permission Target" permission_target_name = params["name"] target = self._prefix + "permissions/" + permission_target_name return self._request("PUT", target, api_method, kwargs, params=params) def delete_permission_target(self, permission_target_name, **kwargs): """ Delete an Artifactory permission target. Parameters ---------- permission_target_name: str Name of the permission target **kwargs Keyword arguments """ api_method = self._category + "Delete Permission Target" target = self._prefix + "permissions/" + permission_target_name return self._request("DELETE", target, api_method, kwargs) def effective_item_permissions(self, repo_key, item_path, **kwargs): """ Return a list of effective permissions. (for the specified item(file or folder). Parameters ---------- repo_key: str Key of the repository item_path: str The path of the item in the repository **kwargs Keyword arguments """ api_method = self._category + "Effective item permissions" target = "storage/" + repo_key + "/" + item_path + "?permissions" return self._request("GET", target, api_method, kwargs) # Unsupported Methods # def security_configuration() # def activate_artifactory_key_encryption() # def deactivate_artifactory_key_encryption() # def get_gpg_public_key() # def set_gpg_public_key() # def set_gpg_private_key() # def set_gpg_pass_phrase() # def create_token() # def refresh_token() # def revoke_token() # def get_service_id() # def get_certificates() # def get_certificate() # def add_certificate() # def delete_certificate() PK!$<+ + rtpy/support.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the SUPPORT REST API Methods category.""" from .tools import RtpyBase class RtpySupport(RtpyBase): """SUPPORT methods category.""" def create_bundle(self, params, **kwargs): """ Create a new support bundle. Parameters ---------- params: dict Settings of the bundle **kwargs Keyword arguments """ api_method = self._category + "Create Bundle" target = self._prefix + "bundles/" return self._request("POST", target, api_method, kwargs, params=params) def list_bundles(self, **kwargs): """ List previously created bundle currently stored in the system. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "List Bundles" target = self._prefix + "bundles/" return self._request("GET", target, api_method, kwargs) def get_bundle(self, bundle_name, **kwargs): """ Download a previously created bundle currently stored in the system. Parameters ---------- bundle_name: str Name of the bundle **kwargs Keyword arguments """ api_method = self._category + "Get Bundle" if bundle_name[:33] == "/artifactory/api/support/bundles/": bundle_name = bundle_name[33:] target = self._prefix + "bundles/" + bundle_name params = {"Content-Type": "application/json"} return self._request( "GET", target, api_method, kwargs, byte_output=True, params=params ) def delete_bundle(self, bundle_name, **kwargs): """ Delete a previously created bundle from the system. Parameters ---------- bundle_name: str Name of the bundle **kwargs Keyword arguments """ api_method = self._category + "Delete Bundle" target = self._prefix + "bundles" + bundle_name return self._request("DELETE", target, api_method, kwargs) PK!ؑ rtpy/system_and_configuration.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the SYSTEM AND CONFIGURATION REST API Methods category.""" from .tools import RtpyBase class RtpySystemAndConfiguration(RtpyBase): """SYSTEM AND CONFIGURATION methods category.""" def system_info(self, **kwargs): """ Get general system information. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "System Info" return self._request("GET", self._prefix, api_method, kwargs) def system_health_ping(self, **kwargs): """ Get a simple status response about the state of Artifactory. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "System Health Ping" target = self._prefix + "ping" return self._request("GET", target, api_method, kwargs) # def verify_connection() def general_configuration(self, **kwargs): """ Get the general configuration (artifactory.config.xml). Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "General Configuration" target = self._prefix + "configuration" return self._request("GET", target, api_method, kwargs) def save_general_configuration(self, xml_file_path, **kwargs): """ Save the general configuration (artifactory.config.xml). Parameters ---------- xml_file_path: str Path of the xml file to POST **kwargs Keyword arguments """ api_method = self._category + "Save General Configuration" target = self._prefix + "configuration" myparams = {"Content-Type": "application/xml"} with open(xml_file_path, "rb") as files: return self._request( "POST", target, api_method, kwargs, params=myparams, data=files ) # Unsupported method # def update_custom_url_base(new_url) def license_information(self, **kwargs): """ Retrieve information about the currently installed license. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Licence Information" target = self._prefix + "license" return self._request("GET", target, api_method, kwargs) def install_license(self, params, **kwargs): """ Install new license key or change the current one. Parameters ---------- params: str Settings of the license **kwargs Keyword arguments """ # The JSON output in case of an error is currently incorrect api_method = self._category + "Install License" target = self._prefix + "license" return self._request("POST", target, api_method, kwargs, params=params) # Unsupported methods # def ha_license_information() # def install_ha_cluster_licenses() # def delete_ha_cluster_license() def version_and_addons_information(self, **kwargs): """ Retrieve information about versions and addons. (the current Artifactory version, revision, and currently installed Add-ons). Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Versions and Add-ons Information" target = self._prefix + "version" return self._request("GET", target, api_method, kwargs) def get_reverse_proxy_configuration(self, **kwargs): """ Retrieve the reverse proxy configuration. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Get Reverse Proxy Configuration" target = self._prefix + "configuration/webServer" return self._request("GET", target, api_method, kwargs) # Unsupported method # def update_reverse_proxy_configuration() def get_reverse_proxy_snippet(self, **kwargs): """ Get the reverse proxy configuration snippet in text format. Parameters ---------- **kwargs Keyword arguments """ api_method = self._category + "Get Reverse Proxy Snippet" target = self._prefix + "configuration/reverseProxy/nginx" return self._request("GET", target, api_method, kwargs) PK!j><= 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 class MalformedAfApiError(HTTPError): """Raised when encountering a malformed Artifactory REST API error JSON.""" pass 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) 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 class UserSettingsError(ValueError): """Raised if some of the provided settings are incorrect or missing.""" pass 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 PK!Qo&& rtpy/xray.py# 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 'LICENSE.md' file # or at 'http://www.apache.org/licenses/LICENSE-2.0'. """Functions for the XRAY REST API Methods category.""" PK!A%G,G,rtpy-1.4.8.dist-info/LICENSE.md Apache License Version 2.0, January 2004 http://www.apache.org/licenses/ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION 1. Definitions. "License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. "Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. "Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. "You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. "Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. "Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. "Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below). "Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof. "Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution." "Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work. 2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. 3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed. 4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: (a) You must give any other recipients of the Work or Derivative Works a copy of this License; and (b) You must cause any modified files to carry prominent notices stating that You changed the files; and (c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and (d) If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License. You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License. 5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions. 6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file. 7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License. 8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages. 9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability. END OF TERMS AND CONDITIONS APPENDIX: How to apply the Apache License to your work. To apply the Apache License to your work, attach the following boilerplate notice, with the fields enclosed by brackets "{}" replaced with your own identifying information. (Don't include the brackets!) The text should be enclosed in the appropriate comment syntax for the file format. We also recommend that a file or class name and description of purpose be included on the same "printed page" as the copyright notice for easier identification within third-party archives. Copyright 2018 Orange Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 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.PK!H|n-WYrtpy-1.4.8.dist-info/WHEEL A н#Z;/" bFF]xzwK;<*mTֻ0*Ri.4Vm0[H, JPK!H6 R rtpy-1.4.8.dist-info/METADATAVmo6_$j8RM;#/[4`-m6ɑ=aQ;[:s<ϸWNh6;&wʃci1%3EA= 9}j,X-1s2sLؔFӊM͕b8 !NAv}3ΗJgȳ1)>, :>[!צi\V/u<negg<29|z.έEj_wykceAd.WVa}f3}znV#Y g$/ump]4[-~horp@7p&SDɏ2,rZQsܛg7)OCE}XO}qoe`*\ȍ~>,By%e]z@%QvG'1y(k_:]˭VRfpfG([X5Cm kΈE=>btW 4W [a8~20$ ?m!- [.+)hBHD// I$%XоZ2ےekO3>F*B>ҿ=:O oJxuSeʥȨ&u5wW_A*@q"v^`iU-/T: W#-@M*`qC .!?gh{-]u:^}G(Cy:SwWf*BoƅF:txL Ya7z`ZA%`~-*q -qYA%ٜzEyxB M+PK!HsԿ4rtpy-1.4.8.dist-info/RECORDuɲH}= 2@. @D&d|s՛\}L7B—f[?sr Y5a|OE4v]8N JØ0%5 -8fȡ:+u =nWub2lԡ>9j1F5]}m=ؔbsOА0MSTPԘ|$KXF1\FCJG./Oա]\s='=IOnXn+ڠъzfPw.4u޺Bɳuzۈ,rc([\*؛"ReD8aQޟ'ÁEᔝ(NplL/R]FYH8+Gh(M<