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

add current files

Browse Source
This commit is contained in:
WorldTeacher
2025-12-06 15:37:31 +01:00
parent 1877905473
commit c55562a019
18 changed files with 570 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/komgapi/endpoints/announcement_controller.py
View File

@@ -1,16 +1,26 @@
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 typing import List, Optional
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class AnnouncementController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getAnnouncements(self) -> List[Announcement]:
url = self.url + "announcements"

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

@@ -1,13 +1,13 @@
import sys
from typing import Any, Optional, Union
import httpx
import loguru
from httpx_retries import Retry, RetryTransport
from komgapi.errors import KomgaError, ResultErrror
from typing import Any, Union
from limit import limit # type:ignore
import loguru
import sys
import json
import time
from komgapi.errors import KomgaError
log = loguru.logger
log.remove()
log.add("logs/komga_api.log", rotation="1 week", retention="1 month")
@@ -17,22 +17,67 @@ log.add(sys.stdout, level="INFO")
class BaseAPI:
def __init__(
self,
username: str,
password: str,
url: str,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
api_version: int = 1,
) -> None:
self._username = username
self._password = password
"""
Initialize the BaseAPI class.
Args:
url (str): Base URL of the API.
username (Optional[str]): Username for basic authentication. Defaults to None.
password (Optional[str]): Password for basic authentication. Defaults to None.
api_key (Optional[str]): API key for token-based authentication. Defaults to None.
timeout (int): Timeout for requests in seconds. Defaults to 20.
api_version (int): API version to use. Defaults to 1.
"""
if isinstance(api_version, int):
api_version = str(api_version)
self.url = url + f"api/v{api_version}/"
self.timeout = timeout
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
self.api_key = api_key
self.username = username
self.password = password
if api_key:
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"X-API-Key": api_key,
}
elif username and password:
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
else:
raise ValueError("Either API key or username/password must be provided.")
def _get_auth(self) -> Optional[httpx.Auth]:
"""
Get the authentication method for the request.
Returns:
Optional[httpx.Auth]: BasicAuth object if username/password is used, None otherwise.
"""
if self.username and self.password:
return httpx.BasicAuth(self.username, self.password)
return None
def setParams(self, locals: dict[Any, Any]) -> dict[Any, Any]:
"""
Filter and return valid parameters for the request.
Args:
locals (dict[Any, Any]): Local variables to filter.
Returns:
dict[Any, Any]: Filtered parameters.
"""
return {
param_name: param
for param_name, param in locals.items()
@@ -41,36 +86,58 @@ class BaseAPI:
}
def test_connection(self):
"""
Test the connection to the API.
Returns:
bool: True if the connection is successful, False otherwise.
"""
try:
with httpx.Client(timeout=self.timeout) as client:
client.get(self.url, headers=self.headers)
with httpx.Client(timeout=self.timeout, headers=self.headers) as client:
client.get(self.url, auth=self._get_auth())
return True
except httpx.RequestError:
return False
def overwriteVersion(self, version: int):
"""
Overwrite the API version in the base URL.
Args:
version (int): New API version.
Returns:
BaseAPI: Updated BaseAPI instance.
"""
self.url = self.url.replace("api/v1/", f"api/v{version}/")
return self
@limit(1, 1)
def getRequest(self, url: str, params: Union[dict[Any, Any], None] = None) -> Any:
"""
Send a GET request to the API.
Args:
url (str): API endpoint URL.
params (Union[dict[Any, Any], None]): Query parameters.
Returns:
Any: JSON response from the server.
"""
if params is None:
params = {}
try:
with httpx.Client(
timeout=self.timeout,
auth=(self._username, self._password),
auth=self._get_auth(),
transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5)),
headers=self.headers,
) as client:
response = client.get(url, params=params, headers=self.headers)
if response.status_code != 200:
return self.getRequest(url, params)
response = client.get(url, params=params)
return response.json()
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
except httpx.RequestError as e:
raise KomgaError(f"Request Error: {e}") from e
def postRequest(
self,
@@ -78,111 +145,96 @@ class BaseAPI:
data: Union[dict[Any, Any], None] = None,
body: Union[dict[Any, Any], None] = None,
):
"""
Send a POST request to the API.
Args:
url (str): API endpoint URL.
data (Union[dict[Any, Any], None]): Query parameters.
body (Union[dict[Any, Any], None]): Request body.
Returns:
Any: JSON response from the server.
"""
if data is None:
data = {}
try:
with httpx.Client(
timeout=self.timeout, auth=(self._username, self._password), transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5))
timeout=self.timeout,
auth=self._get_auth(),
transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5)),
headers=self.headers,
) as client:
response = client.post(
url,
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),
)
status_code = response.status_code
if status_code == 202:
return None
elif status_code == 200:
return response.json()
else:
time.sleep(10)
log.error(
"Unexpected status code {} during POST to {}: {}",
status_code,
url,
response.text,
)
return self.postRequest(url, data, body)
except httpx.ConnectError as e:
raise KomgaError(f"Connection Error: {e}") from e
except httpx.TimeoutException as e:
time.sleep(5) # Wait before retrying
log.error("Timeout during POST to {}: {}", url, str(e))
# raise KomgaError(f"Timeout Error: {e}") from e
response.raise_for_status()
return response.json() if response.content else None
except httpx.RequestError as e:
raise KomgaError(f"Request Error: {e}") from e
def patchRequest(self, url: str, data: Union[dict[Any, Any], None] = None):
"""Send PATCH request to API endpoint.
"""
Send a PATCH request to the API.
Args:
url (str): API endpoint URL
data (Union[dict[Any, Any], None]): Data to send in request body
url (str): API endpoint URL.
data (Union[dict[Any, Any], None]): Request body.
Returns:
dict: JSON response from server
Raises:
KomgaError: For connection/timeout errors
ResultError: For invalid responses
Any: JSON response from the server.
"""
if data is None:
data = {}
try:
log.debug("PATCH request to {} with data: {}", url, json.dumps(data))
with httpx.Client(
timeout=self.timeout,
auth=(self._username, self._password),
auth=self._get_auth(),
headers=self.headers,
) as client:
response = client.patch(url, json=data)
response.raise_for_status()
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
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
except httpx.RequestError as e:
raise KomgaError(f"Request Error: {e}") from e
def deleteRequest(self, url: str):
"""
Send a DELETE request to the API.
Args:
url (str): API endpoint URL.
Returns:
Any: JSON response from the server.
"""
try:
with httpx.Client(
timeout=self.timeout, auth=(self._username, self._password)
timeout=self.timeout,
auth=self._get_auth(),
headers=self.headers,
) as client:
response = client.delete(url, headers=self.headers)
response = client.delete(url)
response.raise_for_status()
return response.json()
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
return response.json() if response.content else None
except httpx.RequestError as e:
raise KomgaError(f"Request Error: {e}") from e
@classmethod
def from_env(cls):
"""
Create a BaseAPI instance using environment variables.
Returns:
BaseAPI: Configured BaseAPI instance.
"""
import os
return cls(
os.environ["KOMGA_USERNAME"],
os.environ["KOMGA_PASSWORD"],
os.environ["KOMGA_URL"],
url=os.environ["KOMGA_URL"],
username=os.environ.get("KOMGA_USERNAME"),
password=os.environ.get("KOMGA_PASSWORD"),
api_key=os.environ.get("KOMGA_API_KEY"),
)

26
src/komgapi/endpoints/book_controller.py
View File

@@ -1,15 +1,29 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
from typing import Any, Dict, List, Optional, Union
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class BookController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
@typing_extensions.deprecated("This function is deprecated.")
def getBooks(

28
src/komgapi/endpoints/claim_controller.py
View File

@@ -1,16 +1,26 @@
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 typing import Optional
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class ClaimController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getClaim(self) -> dict:
url = self.url + "claim"

23
src/komgapi/endpoints/common_controller.py
View File

@@ -1,14 +1,31 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
from typing import Optional
import requests
import typing_extensions
from komgapi.schemas import *
from .baseapi import BaseAPI
class CommonBookController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getBookFile(
self, book_id: str, download_path: str = "~/Downloads"

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

@@ -1,11 +1,26 @@
from .baseapi import BaseAPI
from typing import List, Optional, Dict, Any, Union
from typing import Any, Dict, List, Optional, Union
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class LibraryController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getLibraries(self) -> List[Library]:
url = self.url + "libraries"

27
src/komgapi/endpoints/readlist_controller.py
View File

@@ -1,16 +1,31 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
from typing import Any, Dict, List, Optional, Union
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
from komgapi.errors import KomgaError, LoginError, ResultErrror
from komgapi.errors import KomgaError
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class ReadListController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getReadlists(
self,

48
src/komgapi/endpoints/referential_controller.py
View File

@@ -1,19 +1,24 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
from typing import List
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
from .baseapi import BaseAPI
class ReferentialController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(self, username, password, api_key, url, timeout=20) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getAgeRatings(self, library_id: str, collection_id: str) -> str:
url = self.url + f"age-ratings"
url = self.url + "age-ratings"
data = self.getRequest(url)
return data
@@ -24,24 +29,24 @@ class ReferentialController(BaseAPI):
collection_id: str = None,
series_id: str = None,
) -> List[Author]:
url = self.url + f"authors"
url = self.url + "authors"
params = self.setParams(locals())
data = self.getRequest(url, params)
return [Author(**author) for author in data]
def getAuthorNames(self, search: str = None) -> List[str]:
url = self.url + f"authors/names"
url = self.url + "authors/names"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getAuthorRoles(self) -> List[str]:
url = self.url + f"authors/roles"
url = self.url + "authors/roles"
data = self.getRequest(url)
return data
def getGenres(self, library_id: str = None, collection_id: str = None) -> List[str]:
url = self.url + f"genres"
url = self.url + "genres"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -49,7 +54,7 @@ class ReferentialController(BaseAPI):
def getLanguages(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"languages"
url = self.url + "languages"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -57,7 +62,7 @@ class ReferentialController(BaseAPI):
def getPublishers(
self, search: str = None, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"publishers"
url = self.url + "publishers"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -65,7 +70,7 @@ class ReferentialController(BaseAPI):
def getReleaseDates(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"release-dates"
url = self.url + "release-dates"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -73,19 +78,19 @@ class ReferentialController(BaseAPI):
def getSharingLabels(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"sharing-labels"
url = self.url + "sharing-labels"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getTags(self, library_id: str = None, collection_id: str = None) -> List[str]:
url = self.url + f"tags"
url = self.url + "tags"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getBookTags(self, library_id: str = None, readlist_id: str = None) -> List[str]:
url = self.url + f"book-tags"
url = self.url + "book-tags"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -93,7 +98,7 @@ class ReferentialController(BaseAPI):
def getSeriesTags(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"series-tags"
url = self.url + "series-tags"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
@@ -110,10 +115,9 @@ class ReferentialController(BaseAPI):
page: int = None,
size: int = None,
) -> List[Author]:
url = self.url + f"authors"
url = self.url + "authors"
params = self.setParams(locals())
print(params)
import requests
data: requests.Response = self.overwriteVersion(2).getRequest(url, params)
content = data.json()

26
src/komgapi/endpoints/series_collection_controller.py
View File

@@ -1,16 +1,28 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
from typing import Any, Dict, List, Optional
import requests
import typing_extensions
from typing import List, Optional, Dict, Any
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class SeriesCollectionController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getCollections(
self,

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

@@ -1,14 +1,15 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
import json
from komgapi.schemas import Series, Book, Collection, Thumbnail
import sys
from typing import Any, Dict, List, Optional, Union
import loguru
import sys
import requests
import typing_extensions
from komgapi.schemas import Book, Collection, Error, Series, Thumbnail
from .baseapi import BaseAPI
log = loguru.logger
log.remove()
@@ -17,8 +18,21 @@ log.add(sys.stdout)
class SeriesController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getAllSeries(
self,
@@ -60,7 +74,7 @@ class SeriesController(BaseAPI):
ret.append(Series(**series))
return ret
def getSeries(self, series_id: str) -> Series:
def getSeries(self, series_id: str) -> Union[Series, Error]:
"""Get a single series from the server.
Args:
@@ -71,6 +85,8 @@ class SeriesController(BaseAPI):
"""
url = self.url + f"series/{series_id}"
data = self.getRequest(url)
if "status" in data and "error" in data:
return Error(**data)
return Series(**data)
def analyzeSeries(self, series_id: str) -> Optional[Dict[str, Any]]:

28
src/komgapi/endpoints/settings_controller.py
View File

@@ -1,16 +1,26 @@
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 typing import Optional
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class SettingsController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
)
def getSettings(self) -> Settings:
url = self.url + "settings"

30
src/komgapi/endpoints/user_controller.py
View File

@@ -1,16 +1,28 @@
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 typing import List, Optional
from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class UserController(BaseAPI):
def __init__(self, username, password, url, timeout=20, api_version="2") -> None:
super().__init__(username, password, url, timeout, api_version="2")
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_key: Optional[str] = None,
url: str = "",
timeout: int = 20,
api_version="2",
) -> None:
super().__init__(
username=username,
password=password,
api_key=api_key,
url=url,
timeout=timeout,
api_version=api_version,
)
def getUsers(self) -> List[User]:
url = self.url + "users"