KomSuite/komgAPI
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity

update codebase

Browse Source
This commit is contained in:
WorldTeacher
2025-04-17 10:57:57 +02:00
parent e685c7b930
commit 759c01380f
37 changed files with 378 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/komgapi/endpoints/baseapi.py
View File

@@ -3,6 +3,14 @@ from komgapi.errors import KomgaError, LoginError, ResultErrror
from typing import Any, Union
from limit import limit
import loguru
import sys
log = loguru.logger
log.remove()
log.add("logs/komga_api.log", rotation="1 week", retention="1 month")
log.add(sys.stdout, level="INFO")
class BaseAPI:
def __init__(self, username, password, url, timeout=20, api_version=1) -> None:
@@ -39,15 +47,16 @@ class BaseAPI:
if params is None:
params = {}
try:
# ic(url, params)
response = requests.get(
url,
auth=(self._username, self._password),
params=params,
timeout=self.timeout,
)
response.raise_for_status()
if response.status_code != 200:
self.getRequest(url, params)
# print(response.content)
log.debug(f"Response: {response.content}")
return response.json()
except ConnectionError as e:
message = f"Connection Error: {e}"
@@ -55,22 +64,37 @@ class BaseAPI:
except requests.exceptions.Timeout as e:
raise KomgaError(f"Timeout Error: {e}") from e
def postRequest(self, url, data: Union[dict, None] = None):
def postRequest(self, url, data: Union[dict, None] = None, body: dict = None):
if data is None:
data = {}
try:
response = requests.post(
url,
auth=(self._username, self._password),
json=data,
timeout=self.timeout,
)
if body is not None:
response = requests.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,
)
response.raise_for_status()
status_code = response.status_code
if status_code != 202:
raise ResultErrror(f"Result Error: {response.json()}")
if status_code == 202:
log.debug(f"Response: {response}")
# raise ResultErrror(f"Result Error: {response}")
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
@@ -81,6 +105,7 @@ class BaseAPI:
if data is None:
data = {}
try:
print("patching data", data, url)
response = requests.patch(
url,
auth=(self._username, self._password),
@@ -88,6 +113,10 @@ class BaseAPI:
timeout=self.timeout,
)
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:

5
src/komgapi/endpoints/library_controller.py
View File

@@ -1,10 +1,5 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
from komgapi.errors import KomgaError, LoginError, ResultErrror
from komgapi.schemas import * # Progress, Series

69
src/komgapi/endpoints/series_controller.py
View File

@@ -7,6 +7,14 @@ from typing import List, Optional, Dict, Any, Union
from komgapi.schemas import * # Progress, Series
import loguru
import sys
log = loguru.logger
log.remove()
log.add("logs/komgapi.log", rotation="1 week", retention="1 month")
log.add(sys.stdout)
class SeriesController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
@@ -14,56 +22,39 @@ class SeriesController(BaseAPI):
def getAllSeries(
self,
library_id: List[str] = None,
collection_id: List[str] = None,
status: List[str] = None,
publisher: List[str] = None,
lang: List[str] = None,
genre: List[str] = None,
tag: List[str] = None,
age_rating: List[str] = None,
release_year: List[str] = None,
deleted: bool = None,
complete: bool = None,
unpaged: bool = True,
sort: List[str] = None,
author: List[str] = None,
oneshot: bool = None,
size: int = None,
page: int = None,
body: dict[Any, Any] = {},
) -> list[Series]:
"""Get all series from the server.
By default, this will return all series in the server. You can filter the results by using the parameters.
"""
Get all series from the server that match the query.
Args:
----
- library_id (List[str], optional): The library to be queried. If None, all available libraries will be used. Defaults to None.
- collection_id (List[str], optional): The collection to be queried. Defaults to None.
- status (List[str], optional): The status of the series. Can be: ENDED,ONGOING,ABANDONED,HIATUS. Defaults to None.
- publisher (List[str], optional): Publisher(s) to be searched for. Defaults to None.
- lang (List[str], optional): Language to query for. Uses two-letter codec. Defaults to None.
- genre (List[str], optional): Genre(s) to query for. Defaults to None.
- tag (List[str], optional): Tag(s) to query. Defaults to None.
- age_rating (List[str], optional): A custom age-rating to search for. Needs to be configured manually. Defaults to None.
- release_year (List[str], optional): When the series were released. Defaults to None.
- deleted (bool, optional): If the series is deleted. Defaults to None.
- complete (bool, optional): Turn to true to only search series that are complete. Complete requires TotalBookCount to be set and equal to BookCount. Defaults to None.
- unpaged (bool, optional): Set to False if a single Page of results should be returned. By default, a page contains 20 entries. Defaults to True.
- sort (List[str], optional): Sorting of the returned data. Sort using asc|desc. Multiple sort criteria are supported. Defaults to None.
- author (List[str], optional): Author(s) to include in the query. Defaults to None.
- oneshot (bool, optional): If the series is categorized as oneshot. Defaults to None.
- size (int, optional): The size of the page. Defaults to None.
- page (int, optional): The page to be returned. Defaults to None.
Parameters
----------
unpaged : bool, optional
limits the amount of results. If set to false, 20 entries will be returned, by default True
sort : List[str], optional
sorting parameter, by default None
size : int, optional
How many entries should be returned. Set to None to get all entries, by default None
page : int, optional
If unpaged is False and size is set, this allows the selection of a subset of result, by default None
body : dict, optional
The query that requests the data from the API, by default {}
Returns:
Returns
-------
- list[Series]: a list of all series that match the query. Each series is represented as a Series object.
list[Series]
The result of the query. Each series is represented as a Series object.
"""
params = locals()
body = params.pop("body")
params = self.setParams(params)
url = self.url + "series"
data = self.getRequest(url, params)
url = self.url + "series/list"
data = self.postRequest(url, data=params, body=body)
ret = []
for series in data["content"]:
ret.append(Series(**series))
@@ -250,6 +241,8 @@ class SeriesController(BaseAPI):
"""
url = self.url + f"series/{series_id}/metadata"
data = self.patchRequest(url, changed_metadata)
log.debug("Changed metadata: {}", data)
return data
def refreshMetadata(self, series_id: str) -> Optional[Dict[str, Any]]: