10077017273

Committed 24 Jul 2024 12:54PM UTC coverage: 89.246%. First build

Build # 10077017273

Build Type

Pull #465

github

Committed by

seallard

Commit Message

Fix type hint

Pull Request Pull Request #465: Use abbreviation

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

2141 of 2399 relevant lines covered (89.25%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.91

/trailblazer/clients/slurm_api_client/slurm_api_client.py

import logging
from pydantic import ValidationError
import requests

from trailblazer.clients.slurm_api_client.dto import (
    SlurmJobResponse,
    SlurmJobsResponse,
)
from trailblazer.exc import ResponseDeserializationError, SlurmAPIClientError


LOG = logging.getLogger(__name__)


class SlurmAPIClient:
    def __init__(self, base_url: str, access_token: str, user_name: str) -> None:
        self.base_url = base_url
        self.headers = {"X-SLURM-USER-NAME": user_name, "X-SLURM-USER-TOKEN": access_token}

    def get_job(self, job_id: str) -> SlurmJobResponse:
        endpoint: str = f"{self.base_url}/slurm/v0.0.40/job/{job_id}"
        try:
            response = requests.get(endpoint, headers=self.headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            LOG.error(f"Error getting job {job_id}: {e.response.content}")
            raise SlurmAPIClientError(e)

        try:
            return SlurmJobResponse.model_validate(response.json())
        except ValidationError as e:
            LOG.error(f"Error deserializing job response: {e}")
            raise ResponseDeserializationError(e)

1	import logging	1✔
2	from pydantic import ValidationError	1✔
3	import requests	1✔
4
5	from trailblazer.clients.slurm_api_client.dto import (	1✔
6	SlurmJobResponse,
7	SlurmJobsResponse,
8	)
9	from trailblazer.exc import ResponseDeserializationError, SlurmAPIClientError	1✔
10
11
12	LOG = logging.getLogger(__name__)	1✔
13
14
15	class SlurmAPIClient:	1✔
16	def __init__(self, base_url: str, access_token: str, user_name: str) -> None:	1✔
17	self.base_url = base_url	1✔
18	self.headers = {"X-SLURM-USER-NAME": user_name, "X-SLURM-USER-TOKEN": access_token}	1✔
19
20	def get_job(self, job_id: str) -> SlurmJobResponse:	1✔
21	endpoint: str = f"{self.base_url}/slurm/v0.0.40/job/{job_id}"	1✔
22	try:	1✔
23	response = requests.get(endpoint, headers=self.headers)	1✔
24	response.raise_for_status()	1✔
25	except requests.exceptions.HTTPError as e:	×
26	LOG.error(f"Error getting job {job_id}: {e.response.content}")	×
27	raise SlurmAPIClientError(e)	×
28
29	try:	1✔
30	return SlurmJobResponse.model_validate(response.json())	1✔
31	except ValidationError as e:	×
32	LOG.error(f"Error deserializing job response: {e}")	×
33	raise ResponseDeserializationError(e)	×

Clinical-Genomics / trailblazer / 10077017273

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous