particle-os/debian-forge-composer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix pylint issues

Browse source 
Add timeout to `requests.post()` and `requests.get()` calls to make
pylint happy.

Signed-off-by: Tomáš Hozza <thozza@redhat.com>
This commit is contained in:
Tomáš Hozza 2024-05-28 17:29:52 +02:00 • committed by Achilleas Koutsou
parent 666baff944
commit 71b9c0ad65
1 changed files with 7 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tools/koji-compose.py
View file

@ -130,22 +130,25 @@ class ComposerAPIClient:
resp = requests.post(self.auth_server + "/token", data={ resp = requests.post(self.auth_server + "/token", data={
"grant_type": "refresh_token", "grant_type": "refresh_token",
"refresh_token": self.refresh_token, "refresh_token": self.refresh_token,
}) }, timeout=5)
if resp.status_code != 200: if resp.status_code != 200:
raise RuntimeError(f"failed to refresh token: {resp.text}") raise RuntimeError(f"failed to refresh token: {resp.text}")
return resp.json()["access_token"] return resp.json()["access_token"]
def submit_compose(self, request): def submit_compose(self, request):
return requests.post(self.api_url + "/compose", json=request, return requests.post(self.api_url + "/compose", json=request,
headers={"Authorization": f"Bearer {self.access_token()}"}) headers={"Authorization": f"Bearer {self.access_token()}"},
timeout=5)
def compose_status(self, compose_id): def compose_status(self, compose_id):
return requests.get(self.api_url + f"/composes/{compose_id}", return requests.get(self.api_url + f"/composes/{compose_id}",
headers={"Authorization": f"Bearer {self.access_token()}"}) headers={"Authorization": f"Bearer {self.access_token()}"},
timeout=5)
def compose_log(self, compose_id): def compose_log(self, compose_id):
return requests.get(self.api_url + f"/composes/{compose_id}/logs", return requests.get(self.api_url + f"/composes/{compose_id}/logs",
headers={"Authorization": f"Bearer {self.access_token()}"}) headers={"Authorization": f"Bearer {self.access_token()}"},
timeout=5)
def get_parser(): def get_parser():