21229382305

Committed 21 Jan 2026 11:22PM UTC coverage: 69.613%. Remained the same

Build # 21229382305

Build Type

push

github

Committed by

robertdstein

Commit Message

Update

Run Details

252 of 362 relevant lines covered (69.61%)

1.39 hits per line

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

53.73

/winterapi/base_api.py

"""
Module with the base class for generic API interactions
"""

import json
import logging
import re
from pathlib import Path

import backoff
import requests
from pydantic import BaseModel

logger = logging.getLogger(__name__)

MAX_TIMEOUT = 30.0


class BaseAPI:
    """
    Base class for interacting with the API
    """

    def get_auth(self):
        """
        Get the authentication details.

        :return: Authentication details.
        """
        raise NotImplementedError

    @staticmethod
    def clean_data(data):
        """
        Clean the data for the API.

        :param data: Data to clean.
        :return: Cleaned data.
        """

        if isinstance(data, list):
            convert = json.dumps(
                [
                    x.model_dump(exclude=set(x.__class__.model_computed_fields.keys()))
                    for x in data
                ]
            )
        elif isinstance(data, BaseModel):
            convert = json.dumps(
                [
                    data.model_dump(
                        exclude=set(data.__class__.model_computed_fields.keys())
                    )
                ]
            )
        else:
            err = f"Unrecognised data type {type(data)}"
            logger.error(err)
            raise TypeError(err)

        return convert

    @backoff.on_exception(
        backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
    )
    def get(self, url, auth=None, data=None, **kwargs) -> requests.Response:
        """
        Run a get request.

        :param url: URL to get.
        :param auth: Authentication details.
        :param data: Data to get.
        :param kwargs: additional arguments for API.
        :return: API response.
        """
        if auth is None:
            auth = self.get_auth()

        if data is not None:
            data = self.clean_data(data)

        res = requests.get(
            url, data=data, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
        )

        if res.status_code != 200:
            err = f"API call failed with '{res}: {res.text}'"
            logger.error(err)
            raise ValueError(err)
        return res

    @backoff.on_exception(
        backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
    )
    def post(
        self, url, data: BaseModel | list[BaseModel], auth=None, **kwargs
    ) -> requests.Response:
        """
        Run a post request.

        :param url: URL to post to.
        :param data: Data to post.
        :param auth: Authentication details.
        :param kwargs: additional arguments for API.
        :return: Response.
        """
        if auth is None:
            auth = self.get_auth()

        convert = self.clean_data(data)

        res = requests.post(
            url, data=convert, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
        )

        if res.status_code != 200:
            err = f"API call failed with '{res}: {res.text}'"
            logger.error(err)
            raise ValueError(err)
        return res

    @backoff.on_exception(
        backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
    )
    def delete(self, url, auth=None, **kwargs) -> requests.Response:
        """
        Run a delete request.

        :param url: URL to post to.
        :param auth: Authentication details.
        :param kwargs: additional arguments for API.
        :return: Response.
        """
        if auth is None:
            auth = self.get_auth()

        res = requests.delete(url, auth=auth, params=kwargs, timeout=MAX_TIMEOUT)

        if res.status_code != 200:
            err = f"API call failed with '{res}: {res.text}'"
            logger.error(err)
            raise ValueError(err)
        return res

    @backoff.on_exception(
        backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT * 4
    )
    def get_stream(
        self, url, output_dir: str | Path | None = None, auth=None, data=None, **kwargs
    ) -> tuple[requests.Response, Path]:
        """
        Run a get request.

        :param url: URL to get.
        :param output_dir: Directory to save the output.
        :param auth: Authentication details.
        :param kwargs: additional arguments for API.
        :return: API response.
        """
        if auth is None:
            auth = self.get_auth()

        if data is not None:
            data = self.clean_data(data)

        if output_dir is None:
            output_dir = Path.home()
            logger.warning(f"No output directory specified, using {output_dir}")

        if not isinstance(output_dir, Path):
            output_dir = Path(output_dir)

        if not output_dir.parent.exists():
            output_dir.parent.mkdir(parents=True)

        fname = "winterapi_output.zip"

        with requests.Session() as session:

            with session.get(
                url,
                data=data,
                auth=auth,
                params=kwargs,
                timeout=4.0 * MAX_TIMEOUT,
                stream=True,
            ) as resp:
                header = resp.headers

                if "Content-Disposition" in header.keys():
                    fname = re.findall("filename=(.+)", header["Content-Disposition"])[
                        0
                    ]

                output_path = output_dir.joinpath(fname)

                with open(output_path, "wb") as output_f:
                    for chunk in resp.iter_content(chunk_size=8192):
                        output_f.write(chunk)

        logger.info(f"Downloaded file to {output_path}")

        return resp, output_path

1	"""
2	Module with the base class for generic API interactions
3	"""
4
5	import json	2✔
6	import logging	2✔
7	import re	2✔
8	from pathlib import Path	2✔
9
10	import backoff	2✔
11	import requests	2✔
12	from pydantic import BaseModel	2✔
13
14	logger = logging.getLogger(__name__)	2✔
15
16	MAX_TIMEOUT = 30.0	2✔
17
18
19	class BaseAPI:	2✔
20	"""
21	Base class for interacting with the API
22	"""
23
24	def get_auth(self):	2✔
25	"""
26	Get the authentication details.
27
28	:return: Authentication details.
29	"""
30	raise NotImplementedError
31
32	@staticmethod	2✔
33	def clean_data(data):	2✔
34	"""
35	Clean the data for the API.
36
37	:param data: Data to clean.
38	:return: Cleaned data.
39	"""
40
41	if isinstance(data, list):	2✔
42	convert = json.dumps(	2✔
43	[
44	x.model_dump(exclude=set(x.__class__.model_computed_fields.keys()))
45	for x in data
46	]
47	)
48	elif isinstance(data, BaseModel):	×
49	convert = json.dumps(	×
50	[
51	data.model_dump(
52	exclude=set(data.__class__.model_computed_fields.keys())
53	)
54	]
55	)
56	else:
57	err = f"Unrecognised data type {type(data)}"
58	logger.error(err)
59	raise TypeError(err)
60
61	return convert	2✔
62
63	@backoff.on_exception(	2✔
64	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
65	)
66	def get(self, url, auth=None, data=None, **kwargs) -> requests.Response:	2✔
67	"""
68	Run a get request.
69
70	:param url: URL to get.
71	:param auth: Authentication details.
72	:param data: Data to get.
73	:param kwargs: additional arguments for API.
74	:return: API response.
75	"""
76	if auth is None:	2✔
77	auth = self.get_auth()	2✔
78
79	if data is not None:	2✔
80	data = self.clean_data(data)	×
81
82	res = requests.get(	2✔
83	url, data=data, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
84	)
85
86	if res.status_code != 200:	2✔
87	err = f"API call failed with '{res}: {res.text}'"
88	logger.error(err)
89	raise ValueError(err)
90	return res	2✔
91
92	@backoff.on_exception(	2✔
93	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
94	)
95	def post(	2✔
96	self, url, data: BaseModel \| list[BaseModel], auth=None, **kwargs
97	) -> requests.Response:
98	"""
99	Run a post request.
100
101	:param url: URL to post to.
102	:param data: Data to post.
103	:param auth: Authentication details.
104	:param kwargs: additional arguments for API.
105	:return: Response.
106	"""
107	if auth is None:	2✔
108	auth = self.get_auth()	2✔
109
110	convert = self.clean_data(data)	2✔
111
112	res = requests.post(	2✔
113	url, data=convert, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
114	)
115
116	if res.status_code != 200:	2✔
117	err = f"API call failed with '{res}: {res.text}'"
118	logger.error(err)
119	raise ValueError(err)
120	return res	2✔
121
122	@backoff.on_exception(	2✔
123	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
124	)
125	def delete(self, url, auth=None, **kwargs) -> requests.Response:	2✔
126	"""
127	Run a delete request.
128
129	:param url: URL to post to.
130	:param auth: Authentication details.
131	:param kwargs: additional arguments for API.
132	:return: Response.
133	"""
134	if auth is None:	×
135	auth = self.get_auth()	×
136
137	res = requests.delete(url, auth=auth, params=kwargs, timeout=MAX_TIMEOUT)	×
138
139	if res.status_code != 200:	×
140	err = f"API call failed with '{res}: {res.text}'"
141	logger.error(err)
142	raise ValueError(err)
143	return res	×
144
145	@backoff.on_exception(	2✔
146	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT * 4
147	)
148	def get_stream(	2✔
149	self, url, output_dir: str \| Path \| None = None, auth=None, data=None, **kwargs
150	) -> tuple[requests.Response, Path]:
151	"""
152	Run a get request.
153
154	:param url: URL to get.
155	:param output_dir: Directory to save the output.
156	:param auth: Authentication details.
157	:param kwargs: additional arguments for API.
158	:return: API response.
159	"""
160	if auth is None:	×
161	auth = self.get_auth()	×
162
163	if data is not None:	×
164	data = self.clean_data(data)	×
165
166	if output_dir is None:	×
167	output_dir = Path.home()	×
168	logger.warning(f"No output directory specified, using {output_dir}")	×
169
170	if not isinstance(output_dir, Path):	×
171	output_dir = Path(output_dir)	×
172
173	if not output_dir.parent.exists():	×
174	output_dir.parent.mkdir(parents=True)	×
175
176	fname = "winterapi_output.zip"	×
177
178	with requests.Session() as session:	×
179
180	with session.get(	×
181	url,
182	data=data,
183	auth=auth,
184	params=kwargs,
185	timeout=4.0 * MAX_TIMEOUT,
186	stream=True,
187	) as resp:
188	header = resp.headers	×
189
190	if "Content-Disposition" in header.keys():	×
191	fname = re.findall("filename=(.+)", header["Content-Disposition"])[	×
192	0
193	]
194
195	output_path = output_dir.joinpath(fname)	×
196
197	with open(output_path, "wb") as output_f:	×
198	for chunk in resp.iter_content(chunk_size=8192):	×
199	output_f.write(chunk)	×
200
201	logger.info(f"Downloaded file to {output_path}")	×
202
203	return resp, output_path	×

winter-telescope / winterapi / 21229382305

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