8071617431

Committed 27 Feb 2024 09:31PM UTC coverage: 70.69%. Remained the same

Build # 8071617431

Build Type

push

github

Committed by

web-flow

Commit Message

Add compatibility with new API (#33)

Run Details

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

246 of 348 relevant lines covered (70.69%)

1.41 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.model_computed_fields.keys()))
                    for x in data
                ]
            )
        elif isinstance(data, BaseModel):
            convert = json.dumps(
                [data.model_dump(exclude=set(data.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
    )
    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=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.model_computed_fields.keys()))
45	for x in data
46	]
47	)
48	elif isinstance(data, BaseModel):	×
NEW 49	convert = json.dumps(	×
50	[data.model_dump(exclude=set(data.model_computed_fields.keys()))]
51	)
52	else:
53	err = f"Unrecognised data type {type(data)}"
54	logger.error(err)
55	raise TypeError(err)
56
57	return convert	2✔
58
59	@backoff.on_exception(	2✔
60	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
61	)
62	def get(self, url, auth=None, data=None, **kwargs) -> requests.Response:	2✔
63	"""
64	Run a get request.
65
66	:param url: URL to get.
67	:param auth: Authentication details.
68	:param data: Data to get.
69	:param kwargs: additional arguments for API.
70	:return: API response.
71	"""
72	if auth is None:	2✔
73	auth = self.get_auth()	2✔
74
75	if data is not None:	2✔
76	data = self.clean_data(data)	×
77
78	res = requests.get(	2✔
79	url, data=data, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
80	)
81
82	if res.status_code != 200:	2✔
83	err = f"API call failed with '{res}: {res.text}'"
84	logger.error(err)
85	raise ValueError(err)
86	return res	2✔
87
88	@backoff.on_exception(	2✔
89	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
90	)
91	def post(	2✔
92	self, url, data: BaseModel \| list[BaseModel], auth=None, **kwargs
93	) -> requests.Response:
94	"""
95	Run a post request.
96
97	:param url: URL to post to.
98	:param data: Data to post.
99	:param auth: Authentication details.
100	:param kwargs: additional arguments for API.
101	:return: Response.
102	"""
103	if auth is None:	2✔
104	auth = self.get_auth()	2✔
105
106	convert = self.clean_data(data)	2✔
107
108	res = requests.post(	2✔
109	url, data=convert, auth=auth, params=kwargs, timeout=MAX_TIMEOUT
110	)
111
112	if res.status_code != 200:	2✔
113	err = f"API call failed with '{res}: {res.text}'"
114	logger.error(err)
115	raise ValueError(err)
116	return res	2✔
117
118	@backoff.on_exception(	2✔
119	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
120	)
121	def delete(self, url, auth=None, **kwargs) -> requests.Response:	2✔
122	"""
123	Run a delete request.
124
125	:param url: URL to post to.
126	:param auth: Authentication details.
127	:param kwargs: additional arguments for API.
128	:return: Response.
129	"""
130	if auth is None:	×
131	auth = self.get_auth()	×
132
133	res = requests.delete(url, auth=auth, params=kwargs, timeout=MAX_TIMEOUT)	×
134
135	if res.status_code != 200:	×
136	err = f"API call failed with '{res}: {res.text}'"
137	logger.error(err)
138	raise ValueError(err)
139	return res	×
140
141	@backoff.on_exception(	2✔
142	backoff.expo, requests.exceptions.RequestException, max_time=MAX_TIMEOUT
143	)
144	def get_stream(	2✔
145	self, url, output_dir: str \| Path \| None = None, auth=None, data=None, **kwargs
146	) -> tuple[requests.Response, Path]:
147	"""
148	Run a get request.
149
150	:param url: URL to get.
151	:param output_dir: Directory to save the output.
152	:param auth: Authentication details.
153	:param kwargs: additional arguments for API.
154	:return: API response.
155	"""
156	if auth is None:	×
157	auth = self.get_auth()	×
158
159	if data is not None:	×
160	data = self.clean_data(data)	×
161
162	if output_dir is None:	×
163	output_dir = Path.home()	×
164	logger.warning(f"No output directory specified, using {output_dir}")	×
165
166	if not isinstance(output_dir, Path):	×
167	output_dir = Path(output_dir)	×
168
169	if not output_dir.parent.exists():	×
170	output_dir.parent.mkdir(parents=True)	×
171
172	fname = "winterapi_output.zip"	×
173
174	with requests.Session() as session:	×
175
176	with session.get(	×
177	url,
178	data=data,
179	auth=auth,
180	params=kwargs,
181	timeout=MAX_TIMEOUT,
182	stream=True,
183	) as resp:
184	header = resp.headers	×
185
186	if "Content-Disposition" in header.keys():	×
187	fname = re.findall("filename=(.+)", header["Content-Disposition"])[	×
188	0
189	]
190
191	output_path = output_dir.joinpath(fname)	×
192
193	with open(output_path, "wb") as output_f:	×
194	for chunk in resp.iter_content(chunk_size=8192):	×
195	output_f.write(chunk)	×
196
197	logger.info(f"Downloaded file to {output_path}")	×
198
199	return resp, output_path	×

winter-telescope / winterapi / 8071617431

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