from datetime import datetime import logging from urllib.parse import urljoin from typing import Dict, List from .models.build import Build from .models.build_task import BuildTask import requests 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_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', '') 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']} 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)