140 lines
5.6 KiB
Python
140 lines
5.6 KiB
Python
from datetime import datetime
|
|
import logging
|
|
from urllib.parse import urljoin
|
|
from typing import Dict, List
|
|
|
|
import requests
|
|
|
|
from .models.build import Build
|
|
from .models.build_task import BuildTask
|
|
from .models.build_node_stats import BuildNodeStats
|
|
from .models.build_stat import BuildStat
|
|
from .models.web_node_stats import WebNodeStats
|
|
|
|
|
|
TZ_OFFSET = '+00:00'
|
|
|
|
|
|
class APIclient():
|
|
"""
|
|
client for working with ALBS API
|
|
"""
|
|
|
|
def __init__(self, api_root: str, jwt: str, timeout: int):
|
|
self.api_root = api_root
|
|
self.jwt = jwt
|
|
self.timeout = timeout
|
|
|
|
def get_builds(self, page_num: int = 1) -> List[Build]:
|
|
ep = '/api/v1/builds'
|
|
url = urljoin(self.api_root, ep)
|
|
params = {'pageNumber': page_num}
|
|
headers = {'accept': 'appilication/json'}
|
|
|
|
response = requests.get(
|
|
url, params=params, headers=headers, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
|
|
result = []
|
|
for b in response.json()['builds']:
|
|
try:
|
|
result.append(self._parse_build(b))
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.error("Cant convert build JSON %s to Buildmodel: %s",
|
|
b, err, exc_info=True)
|
|
return result
|
|
|
|
def get_build(self, build_id: int) -> Build:
|
|
ep = f'/api/v1/builds/{build_id}'
|
|
url = urljoin(self.api_root, ep)
|
|
headers = {'accept': 'application/json'}
|
|
response = requests.get(url, headers=headers, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
return self._parse_build(response.json())
|
|
|
|
def __parse_build_node_stats(self, stats: Dict) -> BuildNodeStats:
|
|
logging.debug('raw json: %s', stats)
|
|
|
|
keys = ['build_all', 'build_binaries', 'build_packages', 'build_srpm', 'build_node_task',
|
|
'cas_notarize_artifacts', 'cas_source_authenticate', 'git_checkout', 'upload']
|
|
params = {}
|
|
for k in keys:
|
|
try:
|
|
params[k] = BuildStat(
|
|
start_ts=datetime.fromisoformat(
|
|
stats[k]['start_ts']+TZ_OFFSET) if stats[k]['start_ts'] else None,
|
|
end_ts=datetime.fromisoformat(
|
|
stats[k]['end_ts']+TZ_OFFSET) if stats[k]['end_ts'] else None)
|
|
except KeyError:
|
|
params[k] = BuildStat()
|
|
build_node_stats = BuildNodeStats(**params)
|
|
logging.debug('BuildNodeStats: %s', build_node_stats)
|
|
return build_node_stats
|
|
|
|
def __parse_web_node_stats(self, stats: Dict) -> WebNodeStats:
|
|
keys = ['build_done', 'logs_processing', 'packages_processing']
|
|
params = {}
|
|
logging.debug('raw json: %s', stats)
|
|
for k in keys:
|
|
try:
|
|
params[k] = BuildStat(
|
|
start_ts=datetime.fromisoformat(
|
|
stats[k]['start_ts']+TZ_OFFSET) if stats[k]['start_ts'] else None,
|
|
end_ts=datetime.fromisoformat(
|
|
stats[k]['end_ts']+TZ_OFFSET) if stats[k]['end_ts'] else None)
|
|
except KeyError:
|
|
params[k] = BuildStat()
|
|
web_node_stats = WebNodeStats(**params)
|
|
logging.debug('WebNodeStats %s', web_node_stats)
|
|
return web_node_stats
|
|
|
|
def _parse_build_tasks(self, tasks_json: Dict, build_id: int) -> List[BuildTask]:
|
|
result = []
|
|
for task in tasks_json:
|
|
try:
|
|
started_at = datetime.fromisoformat(
|
|
task['started_at']+TZ_OFFSET) if task['started_at'] else None
|
|
finished_at = datetime.fromisoformat(
|
|
task['finished_at']+TZ_OFFSET) if task['finished_at'] else None
|
|
name = task['ref']['url'].split('/')[-1].replace('.git', '')
|
|
if not task['performance_stats']:
|
|
logging.warning(
|
|
"no perfomance_stats for build_id: %s, build_task_id: %s", build_id, task['id'])
|
|
stats = {'build_node_stats': {}, 'build_done_stats': {}}
|
|
else:
|
|
stats = task['performance_stats'][0]['statistics']
|
|
|
|
params = {'id': task['id'],
|
|
'name': name,
|
|
'build_id': build_id,
|
|
'started_at': started_at,
|
|
'finished_at': finished_at,
|
|
'arch': task['arch'],
|
|
'status_id': task['status'],
|
|
'build_node_stats': self.__parse_build_node_stats(stats['build_node_stats']),
|
|
'web_node_stats': self.__parse_web_node_stats(stats['build_done_stats'])}
|
|
result.append(BuildTask(**params))
|
|
except Exception as err: # pylint: disable=broad-except
|
|
logging.error("Cant convert build_task JSON %s (build_id %s) to BuildTask model: %s",
|
|
task, build_id, err, exc_info=True)
|
|
|
|
result.sort(key=lambda x: x.id, reverse=True)
|
|
return result
|
|
|
|
def _parse_build(self, build_json: Dict) -> Build:
|
|
url = f"https://build.almalinux.org/build/{build_json['id']}"
|
|
created_at = datetime.fromisoformat(build_json['created_at']+TZ_OFFSET)
|
|
finished_at = datetime.fromisoformat(
|
|
build_json['finished_at']+TZ_OFFSET) if build_json['finished_at'] else None
|
|
build_tasks = self._parse_build_tasks(
|
|
build_json['tasks'], build_json['id'])
|
|
|
|
params = {
|
|
'id': build_json['id'],
|
|
'url': url,
|
|
'created_at': created_at,
|
|
'finished_at': finished_at,
|
|
'build_tasks': build_tasks}
|
|
|
|
return Build(**params)
|