refactor: started switch to new api endpoints & naming

dev
Jef Roosens 2022-07-31 13:26:24 +02:00
parent e9a189b309
commit 1f14678997
Signed by: Jef Roosens
GPG Key ID: B75D4F293C7052DB
5 changed files with 55 additions and 15 deletions

View File

@ -18,5 +18,5 @@ lint: venv
'$(VENV)/bin/flake8' '$(SRC_DIR)' '$(VENV)/bin/flake8' '$(SRC_DIR)'
.PHONY: format .PHONY: format
format: venv fmt: venv
'$(VENV)/bin/black' '$(SRC_DIR)' '$(VENV)/bin/black' '$(SRC_DIR)'

View File

@ -1,3 +1,3 @@
from .vieter import Vieter from .vieter import Vieter
__all__ = "Vieter" __all__ = ["Vieter"]

View File

@ -1,8 +0,0 @@
class VieterRepos:
def __init__(self, vieter):
self._vieter = vieter
async def list(self, offset: int = 0, limit: int = 25):
params = {"offset": offset, "limit": limit}
return await self._vieter._get("/api/repos", params=params)

View File

@ -0,0 +1,24 @@
class VieterTargets:
def __init__(self, vieter):
self._vieter = vieter
async def list(self, offset: int = 0, limit: int = 25, repo: str = None) -> dict:
params = {"offset": offset, "limit": limit, "repo": repo}
return await self._vieter._get("/targets", params=params)
async def info(self, repo_id: int) -> dict:
return await self._vieter._get(f"/targets/{repo_id}")
async def create(self, url: str, branch: str, repo: str, schedule: str = None, arch: [str] = None):
params = {
'url': url,
'branch': branch,
'repo': repo,
'schedule': schedule,
}
if arch is not None:
params['arch'] = arch.join(',')
await self._vieter._post('/targets', params=params)

View File

@ -1,9 +1,16 @@
import aiohttp import aiohttp
from .repos import VieterRepos from .targets import VieterTargets
from .logs import VieterBuildLogs from .logs import VieterBuildLogs
API_PREFIX = '/api/v1'
class VieterApiException(Exception):
pass
class Vieter: class Vieter:
def __init__(self, address: str, api_key: str) -> None: def __init__(self, address: str, api_key: str) -> None:
self._address = address self._address = address
@ -12,15 +19,32 @@ class Vieter:
base_url=address, headers={"X-Api-Key": api_key} base_url=address, headers={"X-Api-Key": api_key}
) )
self.repos = VieterRepos(self) self.targets = VieterTargets(self)
self.logs = VieterBuildLogs(self) self.logs = VieterBuildLogs(self)
async def close(self): async def close(self):
await self._client.close() await self._client.close()
async def _do_req(self, method: str, path: str, params: dict = None): async def _do_req(self, method: str, path: str, params: dict = None):
async with self._client.request(method, path, params=params) as res: if params is not None:
return await res.json() params = {k: v for k, v in params.items() if v is not None}
await self._client.request(method, f"{API_PREFIX}{path}", params=params)
async def _do_req_json(self, method: str, path: str, params: dict = None):
if params is not None:
params = {k: v for k, v in params.items() if v is not None}
async with self._client.request(method, f"{API_PREFIX}{path}", params=params) as res:
data = await res.json()
if res.status != 200:
raise VieterApiException(data["message"])
return data["data"]
async def _get(self, *args, **kwargs): async def _get(self, *args, **kwargs):
return await self._do_req("GET", *args, **kwargs) return await self._do_req_json("GET", *args, **kwargs)
async def _post(self, *args, **kwargs):
return await self._do_req("POST", *args, **kwargs)