10060408719

Committed 23 Jul 2024 02:02PM UTC coverage: 89.232%. First build

Build # 10060408719

Build Type

Pull #461

github

Committed by

seallard

Commit Message

Fix module

Pull Request Pull Request #461: Move tower client

Run Details

51 of 80 new or added lines in 3 files covered. (63.75%)

2138 of 2396 relevant lines covered (89.23%)

0.89 hits per line

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

57.97

/trailblazer/clients/tower/tower_client.py

"""Module for Tower Open API."""

import logging
import os
from pathlib import Path

import requests
from requests import ConnectionError, HTTPError
from requests.exceptions import MissingSchema

from trailblazer.clients.tower.models import (
    TowerProcess,
    TowerTask,
    TowerTaskResponse,
    TowerWorkflowResponse,
)
from trailblazer.constants import TOWER_WORKFLOW_STATUS, FileFormat, TrailblazerStatus
from trailblazer.exc import TowerRequirementsError, TrailblazerError
from trailblazer.io.controller import ReadFile

LOG = logging.getLogger(__name__)


class TowerApiClient:
    """A class handling requests and responses to and from the Tower Open APIs.
    Endpoints are defined in https://tower.nf/openapi/."""

    def __init__(self, workflow_id: str):
        self.workflow_id: str = workflow_id
        self.workspace_id: str = os.environ.get("TOWER_WORKSPACE_ID", None)
        self.tower_access_token: str = os.environ.get("TOWER_ACCESS_TOKEN", None)
        self.tower_api_endpoint: str = os.environ.get("TOWER_API_ENDPOINT", None)
        self.workflow_endpoint: str = f"workflow/{self.workflow_id}"
        self.tasks_endpoint: str = f"{self.workflow_endpoint}/tasks"
        self.cancel_endpoint: str = f"{self.workflow_endpoint}/cancel"

    @property
    def headers(self) -> dict:
        """Return headers required for an NF Tower API call.
        Accept and Authorization fields are mandatory."""
        return {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.tower_access_token}",
        }

    @property
    def request_params(self) -> list[tuple]:
        """Return required parameters for an NF Tower API call.
        Workspace ID is mandatory."""
        return [
            ("workspaceId", self.workspace_id),
        ]

    def build_url(self, endpoint: str) -> str:
        """Build an url to query tower."""
        return self.tower_api_endpoint + endpoint

    def send_request(self, url: str) -> dict:
        """Sends a request to the server and returns the response. NF Tower API calls follow the next schema:
        curl -X GET "<URL>?workspaceId=<WORKSPACE_ID>" \
        -H "Accept: application/json"  \
        -H "Authorization: Bearer <TOWER_ACCESS_TOKEN>
        """
        try:
            response = requests.get(
                url,
                headers=self.headers,
                params=self.request_params,
                verify=True,
            )
            if response.status_code == 404:
                LOG.info("Request failed for url %s\n", url)
                response.raise_for_status()
        except (MissingSchema, HTTPError, ConnectionError) as error:
            LOG.info("Request failed for url %s: Error: %s\n", url, error)
            return {}

        return response.json()

    def post_request(self, url: str, data: dict = {}) -> None:
        """Send data via POST request and return response."""
        try:
            response = requests.post(
                url, headers=self.headers, params=self.request_params, json=data
            )
            if response.status_code in {404, 400}:
                LOG.info(f"POST request failed for url {url}\n with message {str(response)}")
                response.raise_for_status()
        except (MissingSchema, HTTPError, ConnectionError) as error:
            LOG.error(f"Request failed for url {url}: Error: {error}\n")
            raise TrailblazerError

    @property
    def meets_requirements(self) -> bool:
        """Return True if required variables are not empty."""
        requirement_map: list[tuple[str, str]] = [
            (self.tower_api_endpoint, "Error: no endpoint specified for Tower Open API request."),
            (
                self.tower_access_token,
                "Error: no access token specified for Tower Open API request.",
            ),
            (self.workspace_id, "Error: no workspace specified for Tower Open API request."),
        ]
        for requirement, error_msg in requirement_map:
            if not requirement:
                LOG.info(error_msg)
                return False
        return True

    @property
    def tasks(self) -> TowerTaskResponse:
        """Return a tasks response with information about submitted jobs."""
        if self.meets_requirements:
            url = self.build_url(endpoint=self.tasks_endpoint)
            return TowerTaskResponse(**self.send_request(url=url))

    @property
    def workflow(self) -> TowerWorkflowResponse:
        """Return a workflow response with general information about the analysis."""
        if self.meets_requirements:
            url = self.build_url(endpoint=self.workflow_endpoint)
            return TowerWorkflowResponse(**self.send_request(url=url))

    def send_cancel_request(self) -> None:
        """Send a POST request to cancel a workflow."""
        if self.meets_requirements:
            url: str = self.build_url(endpoint=self.cancel_endpoint)
            self.post_request(url=url)

1	"""Module for Tower Open API."""
2
3	import logging	1✔
4	import os	1✔
5	from pathlib import Path	1✔
6
7	import requests	1✔
8	from requests import ConnectionError, HTTPError	1✔
9	from requests.exceptions import MissingSchema	1✔
10
11	from trailblazer.clients.tower.models import (	1✔
12	TowerProcess,
13	TowerTask,
14	TowerTaskResponse,
15	TowerWorkflowResponse,
16	)
17	from trailblazer.constants import TOWER_WORKFLOW_STATUS, FileFormat, TrailblazerStatus	1✔
18	from trailblazer.exc import TowerRequirementsError, TrailblazerError	1✔
19	from trailblazer.io.controller import ReadFile	1✔
20
21	LOG = logging.getLogger(__name__)	1✔
22
23
24	class TowerApiClient:	1✔
25	"""A class handling requests and responses to and from the Tower Open APIs.
26	Endpoints are defined in https://tower.nf/openapi/."""
27
28	def __init__(self, workflow_id: str):	1✔
29	self.workflow_id: str = workflow_id	1✔
30	self.workspace_id: str = os.environ.get("TOWER_WORKSPACE_ID", None)	1✔
31	self.tower_access_token: str = os.environ.get("TOWER_ACCESS_TOKEN", None)	1✔
32	self.tower_api_endpoint: str = os.environ.get("TOWER_API_ENDPOINT", None)	1✔
33	self.workflow_endpoint: str = f"workflow/{self.workflow_id}"	1✔
34	self.tasks_endpoint: str = f"{self.workflow_endpoint}/tasks"	1✔
35	self.cancel_endpoint: str = f"{self.workflow_endpoint}/cancel"	1✔
36
37	@property	1✔
38	def headers(self) -> dict:	1✔
39	"""Return headers required for an NF Tower API call.
40	Accept and Authorization fields are mandatory."""
NEW 41	return {	×
42	"Accept": "application/json",
43	"Authorization": f"Bearer {self.tower_access_token}",
44	}
45
46	@property	1✔
47	def request_params(self) -> list[tuple]:	1✔
48	"""Return required parameters for an NF Tower API call.
49	Workspace ID is mandatory."""
NEW 50	return [	×
51	("workspaceId", self.workspace_id),
52	]
53
54	def build_url(self, endpoint: str) -> str:	1✔
55	"""Build an url to query tower."""
NEW 56	return self.tower_api_endpoint + endpoint	×
57
58	def send_request(self, url: str) -> dict:	1✔
59	"""Sends a request to the server and returns the response. NF Tower API calls follow the next schema:
60	curl -X GET "<URL>?workspaceId=<WORKSPACE_ID>" \
61	-H "Accept: application/json" \
62	-H "Authorization: Bearer <TOWER_ACCESS_TOKEN>
63	"""
NEW 64	try:	×
NEW 65	response = requests.get(	×
66	url,
67	headers=self.headers,
68	params=self.request_params,
69	verify=True,
70	)
NEW 71	if response.status_code == 404:	×
NEW 72	LOG.info("Request failed for url %s\n", url)	×
NEW 73	response.raise_for_status()	×
NEW 74	except (MissingSchema, HTTPError, ConnectionError) as error:	×
NEW 75	LOG.info("Request failed for url %s: Error: %s\n", url, error)	×
NEW 76	return {}	×
77
NEW 78	return response.json()	×
79
80	def post_request(self, url: str, data: dict = {}) -> None:	1✔
81	"""Send data via POST request and return response."""
NEW 82	try:	×
NEW 83	response = requests.post(	×
84	url, headers=self.headers, params=self.request_params, json=data
85	)
NEW 86	if response.status_code in {404, 400}:	×
NEW 87	LOG.info(f"POST request failed for url {url}\n with message {str(response)}")	×
NEW 88	response.raise_for_status()	×
NEW 89	except (MissingSchema, HTTPError, ConnectionError) as error:	×
NEW 90	LOG.error(f"Request failed for url {url}: Error: {error}\n")	×
NEW 91	raise TrailblazerError	×
92
93	@property	1✔
94	def meets_requirements(self) -> bool:	1✔
95	"""Return True if required variables are not empty."""
96	requirement_map: list[tuple[str, str]] = [	1✔
97	(self.tower_api_endpoint, "Error: no endpoint specified for Tower Open API request."),
98	(
99	self.tower_access_token,
100	"Error: no access token specified for Tower Open API request.",
101	),
102	(self.workspace_id, "Error: no workspace specified for Tower Open API request."),
103	]
104	for requirement, error_msg in requirement_map:	1✔
105	if not requirement:	1✔
106	LOG.info(error_msg)	1✔
107	return False	1✔
NEW 108	return True	×
109
110	@property	1✔
111	def tasks(self) -> TowerTaskResponse:	1✔
112	"""Return a tasks response with information about submitted jobs."""
NEW 113	if self.meets_requirements:	×
NEW 114	url = self.build_url(endpoint=self.tasks_endpoint)	×
NEW 115	return TowerTaskResponse(**self.send_request(url=url))	×
116
117	@property	1✔
118	def workflow(self) -> TowerWorkflowResponse:	1✔
119	"""Return a workflow response with general information about the analysis."""
120	if self.meets_requirements:	1✔
NEW 121	url = self.build_url(endpoint=self.workflow_endpoint)	×
NEW 122	return TowerWorkflowResponse(**self.send_request(url=url))	×
123
124	def send_cancel_request(self) -> None:	1✔
125	"""Send a POST request to cancel a workflow."""
NEW 126	if self.meets_requirements:	×
NEW 127	url: str = self.build_url(endpoint=self.cancel_endpoint)	×
NEW 128	self.post_request(url=url)	×

Clinical-Genomics / trailblazer / 10060408719

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