Coverage for src/lanraragi_api/api/base.py: 88%
91 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-22 23:20 +0000
1import base64
2from typing import Any, TypeVar, cast
4import requests
5from pydantic import BaseModel, ValidationError
7from lanraragi_api.entity.base import Auth, OperationResponse
8from lanraragi_api.error import (
9 APIHttpError,
10 APIOperationError,
11 APIRequestError,
12 APIResponseDecodeError,
13)
14from lanraragi_api.util.http import merge_headers, merge_params, normalize_params
16T = TypeVar("T", bound=BaseModel)
17R = TypeVar(name="R", bound=OperationResponse)
20class BaseAPICall:
21 """Base class for every API section, handling transport and errors.
23 Subclasses expose one method per endpoint. Those methods build their request
24 through the ``request_*`` helpers below, which send the call and decode the
25 payload into pydantic models.
27 Args:
28 server: Base URL of the LANraragi server, with or without a trailing
29 slash.
30 key: API key sent with every request. Defaults to None, which sends no
31 credentials.
32 auth_way: How the API key is transmitted. Defaults to
33 ``Auth.AUTH_HEADER``.
34 timeout: Timeout applied to every request, either a single value or a
35 ``(connect, read)`` pair. Defaults to None, meaning no timeout.
36 include_error_payload: Whether ``APIOperationError`` carries the raw
37 response payload. Defaults to False.
38 include_operation_error_message: Whether ``APIOperationError`` carries
39 the error message reported by the server. Defaults to True.
40 raise_on_operation_error: Whether a failed operation raises
41 ``APIOperationError`` instead of being returned to the caller.
42 Defaults to False.
43 default_headers: Extra headers sent with every request. Defaults to
44 None, which sends no extra headers.
45 default_params: Extra query parameters sent with every request.
46 Defaults to None, which sends no extra parameters.
48 Raises:
49 APIRequestError: From any endpoint method, when a request cannot be sent
50 to the server because of a connection failure or a timeout.
51 """
53 def __init__(
54 self,
55 server: str,
56 key: str | None = None,
57 auth_way: Auth = Auth.AUTH_HEADER,
58 timeout: float | tuple[int, int] | None = None,
59 include_error_payload: bool = False,
60 include_operation_error_message: bool = True,
61 raise_on_operation_error: bool = False,
62 default_headers: dict[str, str] | None = None,
63 default_params: dict[str, str] | None = None,
64 ):
65 if default_params is None:
66 default_params = {}
67 if default_headers is None:
68 default_headers = {}
70 self.auth_way: Auth = auth_way
71 self.key: str | None = key
72 self.server: str = server
73 self.timeout: float | tuple[int, int] | None = timeout
74 self.include_error_payload: bool = include_error_payload
75 self.include_operation_error_message: bool = include_operation_error_message
76 self.raise_on_operation_error: bool = raise_on_operation_error
77 self.server = self.server.removesuffix("/")
78 self.default_headers: dict[str, str] = dict(default_headers)
79 self.default_params: dict[str, str] = dict(default_params)
81 if self.key is not None:
82 if auth_way == Auth.QUERY_PARAM:
83 self.default_params["key"] = self.key
84 elif auth_way == Auth.AUTH_HEADER:
85 base64_key = base64.b64encode(self.key.encode("utf-8")).decode("utf-8")
86 self.default_headers["Authorization"] = f"Bearer {base64_key}"
88 def _to_url(self, path: str) -> str:
89 """Build the absolute URL of a request from its path.
91 Args:
92 path: Path of the request, with or without a leading slash.
94 Returns:
95 str: Absolute URL of the request.
97 Raises:
98 ValueError: If ``path`` is an absolute URL, or carries a query
99 string or fragment.
100 """
101 if path.startswith(("http://", "https://")):
102 raise ValueError("absolute URLs are not allowed")
103 if "?" in path or "#" in path:
104 raise ValueError("path must not include query or fragment")
105 if not path.startswith("/"):
106 path = f"/{path}"
107 return f"{self.server}{path}"
109 def request(
110 self,
111 method: str,
112 path: str,
113 params: dict[str, Any] | None = None,
114 headers: dict[str, str] | None = None,
115 expected_statuses: set[int] | None = None,
116 timeout: float | tuple[int, int] | None = None,
117 **kwargs: Any,
118 ) -> requests.Response:
119 """Send an HTTP request and return the raw response.
121 Args:
122 method: HTTP method to use.
123 path: Path of the request.
124 params: Query parameters for this request. Defaults to None.
125 headers: Headers for this request. Defaults to None.
126 expected_statuses: Status codes considered a success. Defaults to
127 None, which accepts every 2xx status.
128 timeout: Timeout for this request, overriding the client timeout.
129 Defaults to None, which uses the client timeout.
130 **kwargs: Extra arguments forwarded to ``requests.request``, such as
131 ``data``, ``json``, ``files`` or ``stream``.
133 Returns:
134 requests.Response: Response returned by the server.
136 Raises:
137 APIRequestError: If the request cannot be sent to the server.
138 APIHttpError: If the status code of the response is not part of
139 ``expected_statuses``.
140 """
141 url = self._to_url(path)
143 try:
144 resp = requests.request(
145 method=method.upper(),
146 url=url,
147 params=normalize_params(merge_params(self.default_params, params)),
148 headers=merge_headers(self.default_headers, headers),
149 timeout=self.timeout if timeout is None else timeout,
150 **kwargs,
151 )
152 except requests.RequestException as exc:
153 raise APIRequestError(url, exc.__class__.__name__)
155 if expected_statuses is None:
156 is_ok = 200 <= resp.status_code < 300
157 else:
158 is_ok = resp.status_code in expected_statuses
159 if not is_ok:
160 raise APIHttpError(resp.status_code, url)
162 return resp
164 def parse_json_response(self, response: requests.Response, path: str):
165 """Decode the body of a response as JSON.
167 Args:
168 response: Response to decode.
169 path: Path of the request, used in error messages.
171 Returns:
172 The decoded JSON payload.
174 Raises:
175 APIResponseDecodeError: If the body is not valid JSON.
176 """
177 url = self._to_url(path)
178 try:
179 return response.json()
180 except ValueError as exc:
181 raise APIResponseDecodeError(url, str(exc)) from exc
183 def parse_model(self, model: type[T], payload: Any, path: str) -> T:
184 """Validate a decoded payload against a pydantic model.
186 Args:
187 model: Model to validate the payload with.
188 payload: Decoded JSON payload.
189 path: Path of the request, used in error messages.
191 Returns:
192 BaseModel: Instance of ``model`` built from ``payload``.
194 Raises:
195 APIResponseDecodeError: If ``payload`` does not match ``model``.
196 """
197 url = self._to_url(path)
198 try:
199 return model.model_validate(payload)
200 except ValidationError as exc:
201 raise APIResponseDecodeError(url, str(exc)) from exc
203 def parse_model_list(self, model: type[T], payload: Any, path: str) -> list[T]:
204 """Validate a decoded payload as a list of a pydantic model.
206 Args:
207 model: Model to validate every item of the payload with.
208 payload: Decoded JSON payload, expected to be a list.
209 path: Path of the request, used in error messages.
211 Returns:
212 list[BaseModel]: One instance of ``model`` per item.
214 Raises:
215 APIResponseDecodeError: If ``payload`` is not a list, or if any item
216 does not match ``model``.
217 """
218 if not isinstance(payload, list):
219 raise APIResponseDecodeError(self._to_url(path), "response is not a list")
220 return [
221 self.parse_model(model, item, path) for item in cast(list[Any], payload)
222 ]
224 def request_json(
225 self,
226 method: str,
227 path: str,
228 params: dict[str, Any] | None = None,
229 headers: dict[str, str] | None = None,
230 expected_statuses: set[int] | None = None,
231 timeout: float | tuple[int, int] | None = None,
232 **kwargs: Any,
233 ):
234 """Send a request and decode its body as JSON.
236 Args:
237 method: HTTP method to use.
238 path: Path of the request.
239 params: Query parameters for this request. Defaults to None.
240 headers: Headers for this request. Defaults to None.
241 expected_statuses: Status codes considered a success. Defaults to
242 None, which accepts every 2xx status.
243 timeout: Timeout for this request, overriding the client timeout.
244 Defaults to None, which uses the client timeout.
245 **kwargs: Extra arguments forwarded to ``requests.request``.
247 Returns:
248 The decoded JSON payload.
250 Raises:
251 APIRequestError: If the request cannot be sent to the server.
252 APIHttpError: If the status code of the response is not part of
253 ``expected_statuses``.
254 APIResponseDecodeError: If the body is not valid JSON.
255 """
256 resp = self.request(
257 method=method,
258 path=path,
259 params=params,
260 headers=headers,
261 expected_statuses=expected_statuses,
262 timeout=timeout,
263 **kwargs,
264 )
265 return self.parse_json_response(resp, path)
267 def request_model(
268 self,
269 method: str,
270 path: str,
271 model: type[T],
272 params: dict[str, Any] | None = None,
273 headers: dict[str, str] | None = None,
274 expected_statuses: set[int] | None = None,
275 timeout: float | tuple[int, int] | None = None,
276 **kwargs: Any,
277 ):
278 """Send a request and validate its JSON body against a model.
280 Args:
281 method: HTTP method to use.
282 path: Path of the request.
283 model: Model to validate the body with.
284 params: Query parameters for this request. Defaults to None.
285 headers: Headers for this request. Defaults to None.
286 expected_statuses: Status codes considered a success. Defaults to
287 None, which accepts every 2xx status.
288 timeout: Timeout for this request, overriding the client timeout.
289 Defaults to None, which uses the client timeout.
290 **kwargs: Extra arguments forwarded to ``requests.request``.
292 Returns:
293 BaseModel: Instance of ``model`` built from the response body.
295 Raises:
296 APIRequestError: If the request cannot be sent to the server.
297 APIHttpError: If the status code of the response is not part of
298 ``expected_statuses``.
299 APIResponseDecodeError: If the body is not valid JSON, or does not
300 match ``model``.
301 """
302 payload = self.request_json(
303 method=method,
304 path=path,
305 params=params,
306 headers=headers,
307 expected_statuses=expected_statuses,
308 timeout=timeout,
309 **kwargs,
310 )
311 return self.parse_model(model, payload, path)
313 def request_model_list(
314 self,
315 method: str,
316 path: str,
317 model: type[T],
318 params: dict[str, Any] | None = None,
319 headers: dict[str, str] | None = None,
320 expected_statuses: set[int] | None = None,
321 timeout: float | tuple[int, int] | None = None,
322 **kwargs: Any,
323 ) -> list[T]:
324 """Send a request and validate its JSON body as a list of a model.
326 Args:
327 method: HTTP method to use.
328 path: Path of the request.
329 model: Model to validate every item of the body with.
330 params: Query parameters for this request. Defaults to None.
331 headers: Headers for this request. Defaults to None.
332 expected_statuses: Status codes considered a success. Defaults to
333 None, which accepts every 2xx status.
334 timeout: Timeout for this request, overriding the client timeout.
335 Defaults to None, which uses the client timeout.
336 **kwargs: Extra arguments forwarded to ``requests.request``.
338 Returns:
339 list[BaseModel]: One instance of ``model`` per item of the body.
341 Raises:
342 APIRequestError: If the request cannot be sent to the server.
343 APIHttpError: If the status code of the response is not part of
344 ``expected_statuses``.
345 APIResponseDecodeError: If the body is not a list of objects
346 matching ``model``.
347 """
348 payload = self.request_json(
349 method=method,
350 path=path,
351 params=params,
352 headers=headers,
353 expected_statuses=expected_statuses,
354 timeout=timeout,
355 **kwargs,
356 )
357 return self.parse_model_list(model, payload, path)
359 def request_operation(
360 self,
361 method: str,
362 path: str,
363 model: type[R] = OperationResponse,
364 params: dict[str, Any] | None = None,
365 headers: dict[str, str] | None = None,
366 expected_statuses: set[int] | None = None,
367 raise_on_failure: bool | None = None,
368 timeout: float | tuple[int, int] | None = None,
369 **kwargs: Any,
370 ) -> R:
371 """Send a request whose body reports the result of an operation.
373 The server reports most operation failures inside the body, through the
374 ``success`` field, rather than through the status code. Every status
375 from 200 to 599 is therefore accepted by default, and a failed
376 operation is only turned into an exception when raising is enabled.
378 Args:
379 method: HTTP method to use.
380 path: Path of the request.
381 model: Model to validate the body with. Defaults to
382 ``OperationResponse``.
383 params: Query parameters for this request. Defaults to None.
384 headers: Headers for this request. Defaults to None.
385 expected_statuses: Status codes considered a success. Defaults to
386 None, which accepts every status from 200 to 599.
387 raise_on_failure: Whether to raise on a failed operation, overriding
388 the ``raise_on_operation_error`` setting of the client. Defaults
389 to None, which keeps the setting of the client.
390 timeout: Timeout for this request, overriding the client timeout.
391 Defaults to None, which uses the client timeout.
392 **kwargs: Extra arguments forwarded to ``requests.request``.
394 Returns:
395 OperationResponse: Instance of ``model`` built from the response
396 body.
398 Raises:
399 APIRequestError: If the request cannot be sent to the server.
400 APIHttpError: If the status code of the response is not part of
401 ``expected_statuses``.
402 APIResponseDecodeError: If the body is not valid JSON, or does not
403 match ``model``.
404 APIOperationError: If the operation failed and raising is enabled.
405 """
406 if expected_statuses is None:
407 expected_statuses = set(range(200, 600))
409 resp = self.request(
410 method=method,
411 path=path,
412 params=params,
413 headers=headers,
414 expected_statuses=expected_statuses,
415 timeout=timeout,
416 **kwargs,
417 )
418 payload = self.parse_json_response(resp, path)
419 operation = self.parse_model(model, payload, path)
420 should_raise = self.raise_on_operation_error
421 if raise_on_failure is not None:
422 should_raise = raise_on_failure
424 if should_raise and (resp.status_code >= 400 or operation.success != 1):
425 operation_payload: dict[str, Any] | None = None
426 if self.include_error_payload and isinstance(payload, dict):
427 operation_payload = cast(dict[str, Any], payload)
428 raise APIOperationError(
429 operation.operation,
430 operation.error if self.include_operation_error_message else None,
431 status_code=resp.status_code,
432 payload=operation_payload,
433 )
434 return operation