7796170689

Committed 06 Feb 2024 07:31AM UTC coverage: 70.69%. First build

Build # 7796170689

Build Type

push

github

Committed by

web-flow

Commit Message

Api V2 (#25)

* Update docs

Get schedules working

Update example

Update to allow ToO deletes

Check minimal server version

Add image download

Cleanup

* Add targname

* Finalise new syntax

* Finish API upgrading

* Update tests

* Update tests

Run Details

85 of 162 new or added lines in 3 files covered. (52.47%)

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

winter-telescope / winterapi / 7796170689

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