solidity/scripts/common/rest_api_helpers.py

172 lines
5.6 KiB
Python
Raw Normal View History

from pathlib import Path
from typing import List, Mapping, Optional
import functools
import json
import operator
import shutil
import requests
class APIHelperError(Exception):
pass
class DataUnavailable(APIHelperError):
pass
class InvalidResponse(APIHelperError):
pass
class FileAlreadyExists(APIHelperError):
pass
def query_api(url: str, params: Mapping[str, str], debug_requests=False) -> dict:
if debug_requests:
print(f'REQUEST URL: {url}')
if len(params) > 0:
print(f'QUERY: {params}')
response = requests.get(url, params=params)
response.raise_for_status()
if debug_requests:
json_response = response.json()
print('========== RESPONSE ==========')
if json_response is not None:
print(json.dumps(json_response, indent=4))
else:
print(response.content)
print('==============================')
return response.json()
def download_file(url: str, target_path: Path, overwrite=False):
if not overwrite and target_path.exists():
raise FileAlreadyExists(f"Refusing to overwrite existing file: '{target_path}'.")
with requests.get(url, stream=True) as request:
with open(target_path, 'wb') as target_file:
shutil.copyfileobj(request.raw, target_file)
class Github:
BASE_URL = 'https://api.github.com'
project_slug: str
debug_requests: bool
def __init__(self, project_slug: str, debug_requests: bool):
self.project_slug = project_slug
self.debug_requests = debug_requests
def pull_request(self, pr_id: int) -> dict:
return query_api(
f'{self.BASE_URL}/repos/{self.project_slug}/pulls/{pr_id}',
{},
self.debug_requests
)
class CircleCI:
# None might be a more logical default for max_pages but in most cases we'll actually
# want some limit to prevent flooding the API with requests in case of a bug.
DEFAULT_MAX_PAGES = 10
BASE_URL = 'https://circleci.com/api/v2'
project_slug: str
debug_requests: bool
def __init__(self, project_slug: str, debug_requests: bool):
self.project_slug = project_slug
self.debug_requests = debug_requests
def paginated_query_api_iterator(self, url: str, params: Mapping[str, str], max_pages: int = DEFAULT_MAX_PAGES):
assert 'page-token' not in params
page_count = 0
next_page_token = None
while max_pages is None or page_count < max_pages:
if next_page_token is not None:
params = {**params, 'page-token': next_page_token}
json_response = query_api(url, params, self.debug_requests)
yield json_response['items']
next_page_token = json_response['next_page_token']
page_count += 1
if next_page_token is None:
break
def paginated_query_api(self, url: str, params: Mapping[str, str], max_pages: int = DEFAULT_MAX_PAGES):
return functools.reduce(operator.add, self.paginated_query_api_iterator(url, params, max_pages), [])
def pipelines(
self,
branch: Optional[str] = None,
commit_hash: Optional[str] = None,
excluded_trigger_types: List[str] = None,
) -> List[dict]:
if excluded_trigger_types is None:
excluded_trigger_types = []
for items in self.paginated_query_api_iterator(
f'{self.BASE_URL}/project/gh/{self.project_slug}/pipeline',
{'branch': branch} if branch is not None else {},
max_pages=10,
):
matching_items = [
item
for item in items
if (
(commit_hash is None or item['vcs']['revision'] == commit_hash) and
item['trigger']['type'] not in excluded_trigger_types
)
]
if len(matching_items) > 0:
return matching_items
return []
def workflows(self, pipeline_id: str) -> dict:
return self.paginated_query_api(f'{self.BASE_URL}/pipeline/{pipeline_id}/workflow', {})
def jobs(self, workflow_id: str) -> Mapping[str, dict]:
items = self.paginated_query_api(f'{self.BASE_URL}/workflow/{workflow_id}/job', {})
jobs_by_name = {job['name']: job for job in items}
assert len(jobs_by_name) <= len(items)
if len(jobs_by_name) < len(items):
raise InvalidResponse("Job names in the workflow are not unique.")
return jobs_by_name
def job(self, workflow_id: str, name: str, require_success: bool = False) -> dict:
jobs = self.jobs(workflow_id)
if name not in jobs:
raise DataUnavailable(f"Job {name} is not present in the workflow.")
if require_success and jobs[name]['status'] != 'success':
raise DataUnavailable(
f"Job {name} has failed or is still running. "
f"Current status: {jobs[name]['status']}."
)
return jobs[name]
def artifacts(self, job_number: int) -> Mapping[str, dict]:
items = self.paginated_query_api(f'{self.BASE_URL}/project/gh/{self.project_slug}/{job_number}/artifacts', {})
artifacts_by_name = {artifact['path']: artifact for artifact in items}
assert len(artifacts_by_name) <= len(items)
if len(artifacts_by_name) < len(items):
raise InvalidResponse("Names of artifacts attached to the job are not unique.")
return artifacts_by_name
@staticmethod
def latest_item(items: dict) -> dict:
sorted_items = sorted(items, key=lambda item: item['created_at'], reverse=True)
return sorted_items[0] if len(sorted_items) > 0 else None