Add issue templates, enhance API functionality, and update dependencies
- Introduced bug report and feature request templates for better issue tracking. - Added release workflow for automated versioning and package publishing. - Updated dependencies in `pyproject.toml` for improved functionality. - Refactored API endpoints to use `httpx` for better performance and error handling. - Added new methods in `BookController` and `KOMGAPI_REST` for enhanced book and series management. - Updated schemas to include optional fields for better data handling.
This commit is contained in:
@@ -1,10 +1,12 @@
|
||||
import requests
|
||||
from komgapi.errors import KomgaError, LoginError, ResultErrror
|
||||
import httpx
|
||||
from httpx_retries import Retry, RetryTransport
|
||||
from komgapi.errors import KomgaError, ResultErrror
|
||||
from typing import Any, Union
|
||||
from limit import limit
|
||||
from limit import limit # type:ignore
|
||||
|
||||
import loguru
|
||||
import sys
|
||||
import json
|
||||
|
||||
log = loguru.logger
|
||||
log.remove()
|
||||
@@ -13,29 +15,37 @@ log.add(sys.stdout, level="INFO")
|
||||
|
||||
|
||||
class BaseAPI:
|
||||
def __init__(self, username, password, url, timeout=20, api_version=1) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
username: str,
|
||||
password: str,
|
||||
url: str,
|
||||
timeout: int = 20,
|
||||
api_version: int = 1,
|
||||
) -> None:
|
||||
self._username = username
|
||||
self._password = password
|
||||
self.url = url + f"api/v{api_version}/"
|
||||
self.timeout = timeout
|
||||
self.headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
|
||||
def setParams(self, locals: dict) -> dict:
|
||||
def setParams(self, locals: dict[Any, Any]) -> dict[Any, Any]:
|
||||
return {
|
||||
param_name: param
|
||||
for param_name, param in locals.items()
|
||||
if param is not None and param_name not in ["self", "series_idurl"]
|
||||
if param is not None
|
||||
and param_name not in ["self", "series_idurl", "query", "url"]
|
||||
}
|
||||
|
||||
def test_connection(self):
|
||||
"""Test the connection to the server.
|
||||
|
||||
Returns:
|
||||
bool: True if the connection is successful, False otherwise.
|
||||
"""
|
||||
try:
|
||||
requests.get(self.url, timeout=self.timeout)
|
||||
with httpx.Client(timeout=self.timeout) as client:
|
||||
client.get(self.url, headers=self.headers)
|
||||
return True
|
||||
except requests.exceptions.RequestException:
|
||||
except httpx.RequestError:
|
||||
return False
|
||||
|
||||
def overwriteVersion(self, version: int):
|
||||
@@ -43,126 +53,123 @@ class BaseAPI:
|
||||
return self
|
||||
|
||||
@limit(1, 1)
|
||||
def getRequest(self, url, params: Union[dict, None] = None) -> Any:
|
||||
def getRequest(self, url: str, params: Union[dict[Any, Any], None] = None) -> Any:
|
||||
if params is None:
|
||||
params = {}
|
||||
try:
|
||||
response = requests.get(
|
||||
url,
|
||||
auth=(self._username, self._password),
|
||||
params=params,
|
||||
with httpx.Client(
|
||||
timeout=self.timeout,
|
||||
)
|
||||
auth=(self._username, self._password),
|
||||
transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5)),
|
||||
) as client:
|
||||
response = client.get(url, params=params, headers=self.headers)
|
||||
if response.status_code != 200:
|
||||
self.getRequest(url, params)
|
||||
# print(response.content)
|
||||
log.debug(f"Response: {response.content}")
|
||||
return self.getRequest(url, params)
|
||||
|
||||
return response.json()
|
||||
except ConnectionError as e:
|
||||
message = f"Connection Error: {e}"
|
||||
raise KomgaError(message) from e
|
||||
except requests.exceptions.Timeout as e:
|
||||
except httpx.ConnectError as e:
|
||||
raise KomgaError(f"Connection Error: {e}") from e
|
||||
except httpx.TimeoutException as e:
|
||||
raise KomgaError(f"Timeout Error: {e}") from e
|
||||
|
||||
def postRequest(self, url, data: Union[dict, None] = None, body: dict = None):
|
||||
def postRequest(
|
||||
self,
|
||||
url: str,
|
||||
data: Union[dict[Any, Any], None] = None,
|
||||
body: Union[dict[Any, Any], None] = None,
|
||||
):
|
||||
if data is None:
|
||||
data = {}
|
||||
try:
|
||||
if body is not None:
|
||||
response = requests.post(
|
||||
with httpx.Client(
|
||||
timeout=self.timeout, auth=(self._username, self._password), transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5))
|
||||
) as client:
|
||||
response = client.post(
|
||||
url,
|
||||
auth=(self._username, self._password),
|
||||
json=body,
|
||||
params=data,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
else:
|
||||
response = requests.post(
|
||||
url,
|
||||
auth=(self._username, self._password),
|
||||
json=data,
|
||||
timeout=self.timeout,
|
||||
params=data,
|
||||
json=body if body is not None else {},
|
||||
headers=self.headers,
|
||||
)
|
||||
log.debug(
|
||||
"POST request to {} with data: {}, json: {}",
|
||||
url,
|
||||
json.dumps(data),
|
||||
json.dumps(body),
|
||||
)
|
||||
response.raise_for_status()
|
||||
status_code = response.status_code
|
||||
if status_code == 202:
|
||||
log.debug(f"Response: {response}")
|
||||
# raise ResultErrror(f"Result Error: {response}")
|
||||
return None
|
||||
elif status_code == 200:
|
||||
log.debug(f"Response: {response}")
|
||||
return response.json()
|
||||
else:
|
||||
log.debug(f"Response: {response}")
|
||||
raise ResultErrror(f"Result Error: {response.content}")
|
||||
except ConnectionError as e:
|
||||
message = f"Connection Error: {e}"
|
||||
raise KomgaError(message) from e
|
||||
except requests.exceptions.Timeout as e:
|
||||
except httpx.ConnectError as e:
|
||||
raise KomgaError(f"Connection Error: {e}") from e
|
||||
except httpx.TimeoutException as e:
|
||||
raise KomgaError(f"Timeout Error: {e}") from e
|
||||
|
||||
def patchRequest(self, url, data: Union[dict, None] = None):
|
||||
def patchRequest(self, url: str, data: Union[dict[Any, Any], None] = None):
|
||||
"""Send PATCH request to API endpoint.
|
||||
|
||||
Args:
|
||||
url (str): API endpoint URL
|
||||
data (Union[dict[Any, Any], None]): Data to send in request body
|
||||
|
||||
Returns:
|
||||
dict: JSON response from server
|
||||
|
||||
Raises:
|
||||
KomgaError: For connection/timeout errors
|
||||
ResultError: For invalid responses
|
||||
"""
|
||||
if data is None:
|
||||
data = {}
|
||||
|
||||
try:
|
||||
print("patching data", data, url)
|
||||
response = requests.patch(
|
||||
url,
|
||||
auth=(self._username, self._password),
|
||||
json=data,
|
||||
log.debug("PATCH request to {} with data: {}", url, json.dumps(data))
|
||||
|
||||
with httpx.Client(
|
||||
timeout=self.timeout,
|
||||
)
|
||||
auth=(self._username, self._password),
|
||||
headers=self.headers,
|
||||
) as client:
|
||||
response = client.patch(url, json=data)
|
||||
|
||||
response.raise_for_status()
|
||||
log.debug(
|
||||
f"Response: {response}, {response.status_code}, {response.content}"
|
||||
)
|
||||
print(response.status_code, response.content)
|
||||
if response.status_code != 204:
|
||||
raise ResultErrror(f"Result Error: {response.json()}")
|
||||
except ConnectionError as e:
|
||||
message = f"Connection Error: {e}"
|
||||
raise KomgaError(message) from e
|
||||
except requests.exceptions.Timeout as e:
|
||||
|
||||
if response.status_code == 204:
|
||||
return None
|
||||
|
||||
return response.json() if response.content else None
|
||||
|
||||
except httpx.ConnectError as e:
|
||||
log.error("Connection error during PATCH to {}: {}", url, str(e))
|
||||
raise KomgaError(f"Connection Error: {e}") from e
|
||||
|
||||
except httpx.TimeoutException as e:
|
||||
log.error("Timeout during PATCH to {}: {}", url, str(e))
|
||||
raise KomgaError(f"Timeout Error: {e}") from e
|
||||
|
||||
def putRequest(self, url, data: Union[dict, None] = None):
|
||||
if data is None:
|
||||
data = {}
|
||||
try:
|
||||
response = requests.put(
|
||||
url,
|
||||
auth=(self._username, self._password),
|
||||
json=data,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except ConnectionError as e:
|
||||
message = f"Connection Error: {e}"
|
||||
raise KomgaError(message) from e
|
||||
except requests.exceptions.Timeout as e:
|
||||
raise KomgaError(f"Timeout Error: {e}") from e
|
||||
except httpx.HTTPStatusError as e:
|
||||
log.error("HTTP error during PATCH to {}: {}", url, e.response.text)
|
||||
raise ResultErrror(f"Result Error: {e.response.text}") from e
|
||||
|
||||
def deleteRequest(self, url):
|
||||
def deleteRequest(self, url: str):
|
||||
try:
|
||||
response = requests.delete(
|
||||
url, auth=(self._username, self._password), timeout=self.timeout
|
||||
)
|
||||
with httpx.Client(
|
||||
timeout=self.timeout, auth=(self._username, self._password)
|
||||
) as client:
|
||||
response = client.delete(url, headers=self.headers)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except ConnectionError as e:
|
||||
message = f"Connection Error: {e}"
|
||||
raise KomgaError(message) from e
|
||||
except requests.exceptions.Timeout as e:
|
||||
except httpx.ConnectError as e:
|
||||
raise KomgaError(f"Connection Error: {e}") from e
|
||||
except httpx.TimeoutException as e:
|
||||
raise KomgaError(f"Timeout Error: {e}") from e
|
||||
|
||||
@classmethod
|
||||
def from_env(cls):
|
||||
"""Create a KOMGA API object from environment variables.
|
||||
|
||||
Returns:
|
||||
KOMGAPI_REST: The KOMGA API object.
|
||||
"""
|
||||
import os
|
||||
|
||||
return cls(
|
||||
|
||||
Reference in New Issue
Block a user