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

initial commit

Browse Source
This commit is contained in:
WorldTeacher
2025-02-17 20:28:17 +01:00
commit e685c7b930
53 changed files with 3285 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/komgapi/endpoints/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
from .common_controller import CommonBookController
from .library_controller import LibraryController
from .series_controller import SeriesController
from .readlist_controller import ReadListController
from .series_collection_controller import SeriesCollectionController
from .book_controller import BookController
from .announcement_controller import AnnouncementController
from .user_controller import UserController
from .settings_controller import SettingsController
from .claim_controller import ClaimController
from .referential_controller import ReferentialController

21
src/komgapi/endpoints/announcement_controller.py Normal file
View File

@@ -0,0 +1,21 @@
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
class AnnouncementController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getAnnouncements(self) -> List[Announcement]:
url = self.url + "announcements"
data = self.getRequest(url)
ret = []
for announcement in data:
ret.append(Announcement(**announcement))
return ret

143
src/komgapi/endpoints/baseapi.py Normal file
View File

@@ -0,0 +1,143 @@
import requests
from komgapi.errors import KomgaError, LoginError, ResultErrror
from typing import Any, Union
from limit import limit
class BaseAPI:
def __init__(self, username, password, url, timeout=20, api_version=1) -> None:
self._username = username
self._password = password
self.url = url + f"api/v{api_version}/"
self.timeout = timeout
def setParams(self, locals: dict) -> dict:
return {
param_name: param
for param_name, param in locals.items()
if param is not None and param_name not in ["self", "series_idurl"]
}
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)
return True
except requests.exceptions.RequestException:
return False
def overwriteVersion(self, version: int):
self.url = self.url.replace("api/v1/", f"api/v{version}/")
return self
@limit(1, 1)
def getRequest(self, url, params: Union[dict, None] = None) -> Any:
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()
# print(response.content)
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
def postRequest(self, url, data: Union[dict, None] = None):
if data is None:
data = {}
try:
response = requests.post(
url,
auth=(self._username, self._password),
json=data,
timeout=self.timeout,
)
response.raise_for_status()
status_code = response.status_code
if status_code != 202:
raise ResultErrror(f"Result Error: {response.json()}")
elif status_code == 200:
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
def patchRequest(self, url, data: Union[dict, None] = None):
if data is None:
data = {}
try:
response = requests.patch(
url,
auth=(self._username, self._password),
json=data,
timeout=self.timeout,
)
response.raise_for_status()
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:
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
def deleteRequest(self, url):
try:
response = requests.delete(
url, auth=(self._username, self._password), 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
@classmethod
def from_env(cls):
"""Create a KOMGA API object from environment variables.
Returns:
KOMGAPI_REST: The KOMGA API object.
"""
import os
return cls(
os.environ["KOMGA_USERNAME"],
os.environ["KOMGA_PASSWORD"],
os.environ["KOMGA_URL"],
)

501
src/komgapi/endpoints/book_controller.py Normal file
View File

@@ -0,0 +1,501 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
from komgapi.schemas import * # Progress, Series
class BookController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getBooks(
self,
search_string: str = None,
library_id: List[str] = None,
media_status: List[str] = None,
read_status: List[str] = None,
released_after: List[str] = None,
tag: List[str] = None,
unpaged: bool = True,
page: int = None,
size: int = None,
sort: List[str] = None,
) -> List[Book]:
"""Get all Books that match the given query.
Args:
----
- search_string (str, optional): An optional filter limiting the search to only return results containing the string. Defaults to None.
- library_id (List[str], optional): An optional filter limiting the searched libraries. Defaults to None.
- media_status (List[str], optional): An optional filter limiting the status of the books. Available values : UNKNOWN, ERROR, READY, UNSUPPORTED, OUTDATED. Defaults to None.
- read_status (List[str], optional): An optional filter limiting the read status of the books. Available values : UNREAD, READ, IN_PROGRESS. Defaults to None.
- released_after (List[str], optional): An optional filter limiting the search to books released after the set date. Defaults to None.
- tag (List[str], optional): Optional tags limiting the search. 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.
- page (int, optional): An integer to get a set page. Defaults to None.
- size (int, optional): An integer setting the size of the page. Defaults to None.
- sort (List[str], optional): An option to sort the result. Sorting criteria in the format: property(,asc|desc). Default sort order is ascending. Multiple sort criteria are supported. Defaults to None.
Returns:
-------
List[Book]: A list of all books that match the query. Each book is represented as a Book object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "books"
data = self.getRequest(url, params)
ret = []
for book in data["content"]:
ret.append(Book(**book))
return ret
def getBook(self, book_id: str) -> Book:
"""Get a specific book.
Args:
----
- book_id (str): the ID of the book to get.
Returns:
-------
- Book: the book wrapped in a Book class.
"""
url = self.url + f"books/{book_id}"
data = self.getRequest(url)
return Book(**data)
def analyzeBook(self, book_id: str) -> Optional[Dict[str, Any]]:
"""Analyze a specific book.
Args:
----
- book_id (str): the ID of the book to analyze.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/analyze"
data = self.postRequest(url)
return data
@typing_extensions.deprecated("This function is not implemented yet.")
def deleteBookFile(self, book_id: str) -> Optional[Dict[str, Any]]:
"""Not implemented yet."""
# url = self.url + f"books/{book_id}/file"
# data = self.deleteRequest(url)
raise NotImplementedError
# TODO: add getBookFile (with /*) if this gets a better description
def getManifest(self, book_id: str) -> Manifest:
"""Get the manifest of a book.
Args:
----
- book_id (str): the ID of the book to get the manifest from.
Returns:
-------
- Manifest: the manifest wrapped in a Manifest class.
"""
url = self.url + f"books/{book_id}/manifest"
data = self.getRequest(url)
return Manifest(**data)
def getSpecificManifest(self, book_id: str, manifest: str) -> Manifest:
"""Get the specified Manifest. 3 options are available.
Args:
----
- book_id (str): The ID of the book to get the manifest from.
- manifest (str): the manifest to get. Options are epub, pdf, divina
Returns:
-------
- Manifest: the manifest wrapped in a Manifest class.
"""
url = self.url + f"books/{book_id}/manifest/{manifest}"
data = self.getRequest(url)
return Manifest(**data)
def patchBookMetadata(
self, book_id: str, metadata: BookMetadata
) -> Optional[Dict[str, Any]]:
"""Get the metadata of a book.
Args:
----
- book_id (str): the ID of the book to get the metadata from.
- metadata (BookMetadata): the metadata to set. All fields need to be set.
Returns:
-------
- Metadata: the metadata wrapped in a Metadata class.
"""
url = self.url + f"books/{book_id}/metadata"
data = self.patchRequest(url, metadata.model_dump())
return data
def refreshBookMetadata(self, book_id: str) -> Optional[Dict[str, Any]]:
"""Refresh the metadata of a book.
Args:
----
- book_id (str): the ID of the book to refresh the metadata from.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/metadata/refresh"
data = self.postRequest(url)
return data
def getNextBook(self, book_id: str) -> Book:
"""Get the next Book in the series. If this is not possible, a HTTPError is raised.
Args:
----
- book_id (str): the ID of the book to get the next book from.
Returns:
-------
- Book: the next book wrapped in a Book class.
"""
url = self.url + f"books/{book_id}/next"
data = self.getRequest(url)
return Book(**data)
def getBookPages(self, book_id: str) -> List[Page]:
"""Get the pages of a book.
Args:
----
- book_id (str): the ID of the book to get the pages from.
Returns:
-------
- List[Page]: A list of all pages that match the query. Each page is represented as a Page object.
"""
url = self.url + f"books/{book_id}/pages"
data = self.getRequest(url)
ret = []
for page in data:
ret.append(Page(**page))
return ret
def getBookPage(
self,
book_id: str,
page: int,
convert: str = None,
zero_based: bool = False,
Accept: List[str] = None,
) -> Page:
"""Get a specific page of a book.
Args:
----
- book_id (str): the ID of the book to get the page from.
- page (int): the page to get.
- convert (str, optional): The format to convert the page to. Available values : JPEG, PNG. Defaults to None.
- zero_based (bool, optional): If set to true, pages will start at zero. Defaults to False.
- Accept (List[str]): "Some very limited server driven content negotiation is handled. If a book is a PDF book, and the Accept header contains 'application/pdf' as a more specific type than other 'image/' types, a raw PDF page will be returned."
Returns:
-------
- Page: the page wrapped in a Page class.
"""
url = self.url + f"books/{book_id}/pages/{page}"
data = self.getRequest(url)
return Page(**data)
def getBookPositions(self, book_id: str) -> List[Position]:
"""Get the positions in an ebook. USE UNCLEAR.
Args:
----
- book_id (str): the id of the book to get the positions from.
Returns:
-------
- List[Position]: A list of all positions that match the query. Each position is represented as a Position object.
"""
url = self.url + f"books/{book_id}/positions"
data = self.getRequest(url)
ret = []
for position in data:
ret.append(Position(**position))
return ret
def getPreviousBook(self, book_id: str) -> Book:
"""Get the previous Book in the series. If this is not possible, a HTTPError is raised.
Args:
----
- book_id (str): the ID of the book to get the previous book from.
Returns:
-------
- Book: the previous book wrapped in a Book class.
"""
url = self.url + f"books/{book_id}/previous"
data = self.getRequest(url)
return Book(**data)
def setProgress(self, book_id: str, progress: Progress) -> Optional[Dict[str, Any]]:
"""Set the progress of a book.
Args:
----
- book_id (str): the ID of the book to set the progress to.
- progress (Progress): the progress to set.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/progression"
data = self.putRequest(url, progress.model_dump())
return data
def deleteProgress(self, book_id: str) -> Optional[Dict[str, Any]]:
"""Set the book as unread.
Args:
----
- book_id (str): the ID of the book to delete the progress from.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/read-progress"
def setProgress(
self, book_id: str, progress: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""Set the progress of a book.
Args:
----
- book_id (str): the ID of the book to set the progress to.
- progress (Progress): the progress to set. Needs to be a dict in the format: {"page": int, "completed": bool}
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/read-progress"
data = self.putRequest(url, progress)
return data
def getBookReadlists(self, book_id: str) -> List[Readlist]:
"""Get the readlists of a book.
Args:
----
- book_id (str): the ID of the book to get the readlists from.
Returns:
-------
- List[Readlist]: A list of all readlists that match the query. Each readlist is represented as a Readlist object.
"""
url = self.url + f"books/{book_id}/readlists"
data = self.getRequest(url)
ret = []
for readlist in data:
ret.append(Readlist(**readlist))
return ret
def getBookThumbnail(self, book_id: str) -> bytes:
"""Get the thumbnail of a book.
Args:
----
- book_id (str): the ID of the book to get the thumbnail from.
Returns:
-------
- bytes: the thumbnail as a bytestring.
"""
url = self.url + f"books/{book_id}/thumbnail"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def getBookThumbnails(self, book_id: str) -> List[Thumbnail]:
"""Get the metadata of all available thumbnails for the book.
Args:
----
- book_id (str): the ID of the book to get the thumbnails from.
Returns:
-------
- List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class.
"""
url = self.url + f"books/{book_id}/thumbnails"
data = self.getRequest(url)
ret = []
for thumb in data:
ret.append(Thumbnail(**thumb))
return ret
def setBookThumbnail(
self, book_id: str, selected: bool, file: bytes
) -> Union[Optional[Dict[str, Any]], Thumbnail]:
"""Set a thumbnail based on a bytestring sent through this function.
Args:
----
- book_id (str): the ID of the book to set the thumbnail to.
- selected (bool): Whether the thumbnail should be selected.Set to True if the thumbnail should be selected.
- file (bytes): the thumbnail to set as a bytestring.
Returns:
-------
- Union[Optional[Dict[str, Any]],Thumbnail]: If successful, the server will return the thumbnail data wrapped in a Thumbnail class. If a violation is found, the server will return a dictionary with the violation.
"""
url = self.url + f"books/{book_id}/thumbnails"
params = {"selected": selected, "thumbnail": file}
data = self.postRequest(url, params)
return Thumbnail(**data)
def getSpecificBookThumbnail(self, book_id: str, thumbnail_id: str) -> bytes:
"""Get a specific thumbnail based on the book id and the thumbnail id.
Args:
----
- book_id (str): the ID of the book to get the thumbnail from.
- thumbnail_id (str): the ID of the thumbnail to get.1
Returns:
-------
- bytes: the thumbnail as a bytestring.
"""
url = self.url + f"books/{book_id}/thumbnails/{thumbnail_id}"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def deleteSpecificBookThumbnail(
self, book_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
"""Delete a specific Thumbnail based on book id and thumbnail id.
Args:
----
- book_id (str): the ID of the book to delete the thumbnail from.
- thumbnail_id (str): the ID of the thumbnail to delete.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/thumbnails/{thumbnail_id}"
data = self.deleteRequest(url)
return data
def setSpecificBookThumbnail(
self, book_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
"""Set the thumbnail to be selected.
Args:
----
- book_id (str): the ID of the book to set the thumbnail to.
- thumbnail_id (str): the ID of the thumbnail to set.
Returns:
-------
- Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"books/{book_id}/thumbnail/{thumbnail_id}/selected"
data = self.putRequest(url)
return data
def getDuplicates(
self,
unpaged: bool = True,
page: int = None,
size: int = None,
sort: List[str] = None,
) -> List[Duplicate]:
"""Get a list of duplicates.
Args:
----
- unpaged (bool, optional): If the result should be unpaged. Defaults to True.
- page (int, optional): the page to request. Defaults to None.
- size (int, optional): the size of a page. Defaults to None.
- sort (List[str], optional): Sorting criteria in the format: property(,asc|desc). Default sort order is ascending. Multiple sort criteria are supported.. Defaults to None.
Returns:
-------
- List[Duplicate]: A list of all duplicates that match the query. Each duplicate is represented as a Duplicate object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "books/duplicates"
data = self.getRequest(url, params)
ret = []
for duplicate in data["content"]:
ret.append(Duplicate(**duplicate))
return ret
@typing_extensions.deprecated("This function is not implemented yet.")
def importBook(self):
raise NotImplementedError
def getLatestBooks(
self, unpaged: str = False, page: int = None, size: int = None
) -> List[Book]:
"""Get a list of the newly added or updated books. By default, it returns up to 20 entries.
Setting unpaged to True will return all entries.
Args:
----
- unpaged (str, optional): If the result should be paged. Defaults to False.
- page (int, optional): The page to request. Defaults to None.
- size (int, optional): The size of the page. Defaults to None.
Returns:
-------
- List[Book]: A list of all books that match the query. Each book is represented as a Book object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "books/latest"
data = self.getRequest(url, params)
ret = []
for book in data["content"]:
ret.append(Book(**book))
return ret
def getOnDeckBooks(
self, library_id: List[str] = None, page: int = None, size: int = None
) -> List[Book]:
"""Return first unread book of series with at least one book read and no books in progress..
Args:
----
- library_id (List[str], optional): Optional Filter to set specific libraries. Defaults to None.
- page (int, optional): _description_. Defaults to None.
- size (int, optional): _description_. Defaults to None.
Returns:
-------
- List[Book]: A list of all books that match the query. Each book is represented as a Book object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "books/ondeck"
data = self.getRequest(url, params)
ret = []
for book in data["content"]:
ret.append(Book(**book))
return ret

18
src/komgapi/endpoints/claim_controller.py Normal file
View File

@@ -0,0 +1,18 @@
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
class ClaimController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getClaim(self) -> dict:
url = self.url + "claim"
data = self.getRequest(url)
return data

91
src/komgapi/endpoints/common_controller.py Normal file
View File

@@ -0,0 +1,91 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from komgapi.schemas import *
class CommonBookController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getBookFile(
self, book_id: str, download_path: str = "~/Downloads"
) -> pathlib.Path:
"""Download the book file.
Args:
----
- book_id (str): the ID of the book to download.
- download_path (str, optional): The path to download the file to. Defaults to "~/Downloads".
Returns:
-------
- pathlib.Path: The path to the downloaded file.
"""
url = self.url + f"books/{book_id}/file"
download_path = pathlib.Path(download_path).expanduser()
subprocess.run(
[
"curl",
"-u",
f"{self._username}:{self._password}",
"-o",
f"{download_path}/{book_id}.zip",
url,
]
)
return pathlib.Path(f"{download_path}/{book_id}.zip")
def getRawPage(self, book_id: str, page: int) -> bytes:
"""Get the raw page of a book.
Args:
----
- book_id (str): the ID of the book to get the page from.
- page (int): the page to get.
Returns:
-------
- bytes: the page as a bytestring.
"""
url = self.url + f"books/{book_id}/pages/{page}/raw"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def getPageThumbnail(self, book_id: str, page: int) -> bytes:
"""Get the thumbnail of a page.
Args:
----
- book_id (str): the ID of the book to get the page from.
- page (int): the page to get.
Returns:
-------
- bytes: the thumbnail as a bytestring.
"""
url = self.url + f"books/{book_id}/pages/{page}/thumbnail"
ic(url)
data = requests.get(url, auth=(self._username, self._password))
return data.content
def getProgress(self, book_id: str) -> Progress:
"""Get the progress of a book. The progress contains a timestamp of the latest modification, the device where the modification was made (if the data is present), and a Locator object.
Args:
----
- book_id (str): the ID of the book to get the progress from.
Returns:
-------
- Progress: the progress wrapped in a Progress class.
"""
url = self.url + f"books/{book_id}/progression"
data = self.getRequest(url)
return Progress(**data)
@typing_extensions.deprecated("This function is not implemented yet.")
def getbookRessource(self, book_id: str, ressource: str) -> bytes:
raise NotImplementedError

58
src/komgapi/endpoints/library_controller.py Normal file
View File

@@ -0,0 +1,58 @@
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
class LibraryController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getLibraries(self) -> List[Library]:
url = self.url + "libraries"
data = self.getRequest(url)
ret = []
for library in data:
ret.append(Library(**library))
return ret
def createLibrary(
self, library: CreateLibrary
) -> Union[Library, Optional[Dict[str, Any]]]:
url = self.url + "libraries"
data = self.postRequest(url, library.model_dump())
return Library(**data) if data else None
def getLibrary(self, library_id: str) -> Library:
url = self.url + f"libraries/{library_id}"
data = self.getRequest(url)
return Library(**data)
def patchLibrary(self, library_id: str, changed_library: Dict[str, Any]) -> None:
url = self.url + f"libraries/{library_id}"
data = self.patchRequest(url, changed_library)
return data
def analyzeLibrary(self, library_id: str) -> Optional[Dict[str, Any]]:
url = self.url + f"libraries/{library_id}/analyze"
data = self.postRequest(url)
return data
def emptyTrash(self, library_id: str) -> None:
url = self.url + f"libraries/{library_id}/empty-trash"
data = self.postRequest(url)
return data
def refreshMetadata(self, library_id: str) -> None:
url = self.url + f"libraries/{library_id}/metadata/refresh"
data = self.postRequest(url)
return data
def scanLibrary(self, library_id: str) -> Optional[Dict[str, Any]]:
url = self.url + f"libraries/{library_id}/scan"
data = self.postRequest(url)
return data

256
src/komgapi/endpoints/readlist_controller.py Normal file
View File

@@ -0,0 +1,256 @@
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
class ReadListController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getReadlists(
self,
search_string: str = None,
library_id: List[str] = None,
unpaged: bool = True,
page: int = None,
size: int = None,
) -> List[Series]:
"""Get a list of all readlists.
Args:
search_string (str, optional): The string to search for. Defaults to None.
library_id (List[str], optional): The library to search in. Defaults to None.
unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True.
page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None.
size (int, optional): Defines the size of the page. Defaults to None.
Returns:
List[dict]: A list of all readlists.
"""
params = locals()
params = self.setParams(params)
url = self.url + "readlists"
data = self.getRequest(url, params)
ret = []
for readlist in data["content"]:
ic(readlist)
ret.append(Series(**readlist))
return ret
def setReadlist(
self, readlist: Readlist
) -> Union[Readlist, Optional[Dict[str, Any]]]:
"""Create a new readlist.
Args:
readlist (Readlist): The readlist to create.
Returns:
Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + "readlists"
data = self.postRequest(url, readlist.model_dump())
return Readlist(**data) if data else None
def getReadlist(self, readlist_id: str) -> Series:
"""Get a specific readlist.
Args:
readlist_id (str): The ID of the readlist to get.
Returns:
Series: The readlist wrapped in a Series class.
"""
url = self.url + f"readlists/{readlist_id}"
data = self.getRequest(url)
return Series(**data)
def deleteReadlist(self, readlist_id: str) -> Optional[Dict[str, Any]]:
"""Delete a specific readlist.
Args:
readlist_id (str): The ID of the readlist to delete.
Returns:
Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"readlists/{readlist_id}"
data = self.deleteRequest(url)
return data
def patchReadlist(
self, readlist_id: str, changed_readlist: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""Change the metadata of a readlist.
Args:
readlist_id (str): The ID of the readlist to change.
changed_readlist (Dict[str,Any]): The changed readlist. This should be a dictionary with the changed fields. Only the fields that are changed need to be included.
Returns:
Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"readlists/{readlist_id}"
data = self.patchRequest(url, changed_readlist)
return data
def getReadlistBooks(
self,
readlist_id: str,
library_id: List[str] = None,
read_status: List[str] = None,
media_status: List[str] = None,
deleted: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
author: List[str] = None,
) -> List[Book]:
url = self.url + f"readlists/{readlist_id}/books"
params = locals()
params = self.setParams(params)
data = self.getRequest(url, params)
ret = []
for book in data["content"]:
ret.append(Book(**book))
return ret
def getNextReadlistBook(self, readlist_id: str, book_id: str) -> Book:
"""Get the next book in a readlist.
Args:
readlist_id (str): The ID of the readlist to get the next book from.
book_id (str): The ID of the current book.
Returns:
Book: The next book wrapped in a Book class.
"""
url = self.url + f"readlists/{readlist_id}/books/{book_id}/next"
data = self.getRequest(url)
return Book(**data)
def getPreviousReadlistBook(self, readlist_id: str, book_id: str) -> Book:
"""Get the previous book in a readlist.
Args:
readlist_id (str): The ID of the readlist to get the previous book from.
book_id (str): The ID of the current book.
Returns:
Book: The previous book wrapped in a Book class.
"""
url = self.url + f"readlists/{readlist_id}/books/{book_id}/previous"
data = self.getRequest(url)
return Book(**data)
def getReadlistFile(
self, readlist_id: str, download_path: str = "~/Downloads"
) -> pathlib.Path:
"""Download the complete readlist as a giant zip file.
Unlike the other functions, this one has no timeout. It will wait until the file is downloaded.
WARNING: THIS WILL TAKE A VERY LONG TIME FOR LARGE READLISTS. USE WITH CAUTION.
My test: 800mb took ~7min to download.
Args:
readlist_id (str): ID of the readlist to download the file from.
download_path (str, optional): The path to download the file to. Defaults to "~/Downloads".
Returns:
pathlib.Path: The path to the downloaded file.
"""
url = self.url + f"readlists/{readlist_id}/file"
download_path = pathlib.Path("~/Downloads").expanduser()
subprocess.run(
[
"curl",
"-u",
f"{self._username}:{self._password}",
"-o",
f"{download_path}/{readlist_id}.zip",
url,
]
)
return pathlib.Path(f"{download_path}/{readlist_id}.zip")
def getReadlistThumbnail(self, readlist_id: str) -> str:
"""Get the current active thumbnail of a readlist.
Args:
readlist_id (str): the ID of the readlist to get the thumbnail from.
Returns:
str: the thumbnail of the readlist formatted as bytestring.
"""
url = self.url + f"readlists/{readlist_id}/thumbnail"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def getReadlistThumbnails(self, readlist_id: str) -> List[ReadlistThumbnail]:
"""Get the metadata of all available thumbnails for the readlist.
Args:
readlist_id (str): the ID of the readlist to get the thumbnails from.
Returns:
List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class.
"""
url = self.url + f"readlists/{readlist_id}/thumbnails"
data = self.getRequest(url)
ret = []
for thumb in data:
ret.append(ReadlistThumbnail(**thumb))
return ret
def setReadlistThumbnail(
self, readlist_id: str, selected: bool, thumbnail: bytes
) -> Union[Thumbnail, Optional[Dict[str, Any]]]:
"""Set a new thumbnail for the readlist.
Args:
readlist_id (str): The ID of the readlist to set the thumbnail to.
selected (bool): If the thumbnail should be selected.
thumbnail (bytes): The thumbnail to set.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"readlists/{readlist_id}/thumbnail"
data = self.postRequest(url, {"selected": selected, "thumbnail": thumbnail})
return Thumbnail(**data) if data else None
def getReadlistSpecificThumbnail(
self, readlist_id: str, thumbnail_id: str
) -> bytes:
url = self.url + f"readlists/{readlist_id}/thumbnails/{thumbnail_id}"
try:
data = requests.get(
url, auth=(self._username, self._password), timeout=self.timeout
)
return data.content
except requests.exceptions.Timeout as e:
raise KomgaError(f"Timeout Error: {e}") from e
def deleteReadlistSpecificThumbnail(
self, readlist_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
"""Delete a specific thumbnail of a readlist.
Args:
readlist_id (str): The ID of the readlist to delete the thumbnail from.
thumbnail_id (str): The ID of the thumbnail to delete.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"readlists/{readlist_id}/thumbnails/{thumbnail_id}"
data = self.deleteRequest(url)
return data
def setReadlistSpecificThumbnail(self, readlist_id: str, thumbnail_id: str):
url = self.url + f"readlists/{readlist_id}/thumbnail/{thumbnail_id}/selected"
response = self.putRequest(url)
return response

120
src/komgapi/endpoints/referential_controller.py Normal file
View File

@@ -0,0 +1,120 @@
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
class ReferentialController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getAgeRatings(self, library_id: str, collection_id: str) -> str:
url = self.url + f"age-ratings"
data = self.getRequest(url)
return data
def getAuthors(
self,
search: str = None,
library_id: str = None,
collection_id: str = None,
series_id: str = None,
) -> List[Author]:
url = self.url + f"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"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getAuthorRoles(self) -> List[str]:
url = self.url + f"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"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getLanguages(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"languages"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getPublishers(
self, search: str = None, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"publishers"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getReleaseDates(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"release-dates"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getSharingLabels(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"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"
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"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getSeriesTags(
self, library_id: str = None, collection_id: str = None
) -> List[str]:
url = self.url + f"series-tags"
params = self.setParams(locals())
data = self.getRequest(url, params)
return data
def getAuthorsV2(
self,
search: str = None,
role: str = None,
library_id: str = None,
collection_id: str = None,
series_id: str = None,
readlist_id: str = None,
unpaged: bool = None,
page: int = None,
size: int = None,
) -> List[Author]:
url = self.url + f"authors"
params = self.setParams(locals())
print(params)
import requests
data: requests.Response = self.overwriteVersion(2).getRequest(url, params)
content = data.json()
return [Author(**author) for author in content]

136
src/komgapi/endpoints/series_collection_controller.py Normal file
View File

@@ -0,0 +1,136 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from typing import List, Optional, Dict, Any
from komgapi.schemas import * # Progress, Series
class SeriesCollectionController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getCollections(
self,
search_string: str = None,
library_id: List[str] = None,
unpaged: bool = True,
page: int = None,
size: int = None,
) -> List[Collection]:
url = self.url + "collections"
params = locals()
params = self.setParams(params)
data = self.getRequest(url, params)
ret = []
for collection in data["content"]:
ret.append(Collection(**collection))
return ret
def createCollection(self, collection: dict) -> Optional[Dict[str, Any]]:
"""Create a new collection.
Args:
collection (dict): A dictionary with the collection data.Needs to contain this:
{
"name": string,
"ordered": bool,
"seriesIds": [
string
]
}
Returns:
Optional[Dict[str, Any]]: An optional dictionary with the created collection.
"""
url = self.url + "collections"
data = self.postRequest(url, collection)
return data
def getCollection(self, collection_id: str) -> Collection:
url = self.url + f"collections/{collection_id}"
data = self.getRequest(url)
return Collection(**data)
def deleteCollection(self, collection_id: str) -> Optional[Dict[str, Any]]:
url = self.url + f"collections/{collection_id}"
data = self.deleteRequest(url)
return data
def patchCollection(
self, collection_id: str, changed_collection: dict
) -> Optional[Dict[str, Any]]:
url = self.url + f"collections/{collection_id}"
data = self.patchRequest(url, changed_collection)
return data
def getCollectionSeries(
self,
collection_id: str,
library_id: List[str] = None,
read_status: List[str] = None,
publisher: List[str] = None,
language: List[str] = None,
genre: List[str] = None,
tag: List[str] = None,
age_rating: List[str] = None,
deleted: bool = None,
complete: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
author: List[str] = None,
) -> List[Series]:
params = locals()
params = self.setParams(params)
url = self.url + f"collections/{collection_id}/series"
data = self.getRequest(url, params)
ret = []
for series in data["content"]:
ret.append(Series(**series))
return ret
def getCollectionThumbnail(self, collection_id: str) -> str:
url = self.url + f"collections/{collection_id}/thumbnail"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def getCollectionThumbnails(self, collection_id: str) -> List[Thumbnail]:
url = self.url + f"collections/{collection_id}/thumbnails"
data = self.getRequest(url)
ret = []
for thumb in data:
ret.append(Thumbnail(**thumb))
return ret
def setCollectionThumbnail(
self, collection_id: str, thumbnail_id: str, selected: bool
) -> Optional[Dict[str, Any]]:
url = self.url + f"collections/{collection_id}/thumbnail/{thumbnail_id}"
data = self.putRequest(url, {"selected": selected})
return data
def getSpecificCollectionThumbnail(
self, collection_id: str, thumbnail_id: str
) -> bytes:
url = self.url + f"collections/{collection_id}/thumbnails/{thumbnail_id}"
data = requests.get(url, auth=(self._username, self._password))
return data.content
def deleteSpecificCollectionThumbnail(
self, collection_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
url = self.url + f"collections/{collection_id}/thumbnails/{thumbnail_id}"
data = self.deleteRequest(url)
return data
def setSpecificCollectionThumbnail(
self, collection_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
url = (
self.url + f"collections/{collection_id}/thumbnail/{thumbnail_id}/selected"
)
data = self.putRequest(url)
return data

530
src/komgapi/endpoints/series_controller.py Normal file
View File

@@ -0,0 +1,530 @@
from .baseapi import BaseAPI
import pathlib
import subprocess
import requests
import typing_extensions
from typing import List, Optional, Dict, Any, Union
from komgapi.schemas import * # Progress, Series
class SeriesController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
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,
) -> 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.
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.
Returns:
-------
- list[Series]: a list of all series that match the query. Each series is represented as a Series object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "series"
data = self.getRequest(url, params)
ret = []
for series in data["content"]:
ret.append(Series(**series))
return ret
def getSeries(self, series_id: str) -> Series:
"""Get a single series from the server.
Args:
series_id (str): The ID of the series to get.
Returns:
Series: the series that matches the ID wrapped in a Series class.
"""
url = self.url + f"series/{series_id}"
data = self.getRequest(url)
return Series(**data)
def analyzeSeries(self, series_id: str) -> Optional[Dict[str, Any]]:
"""Instruct the server to analyze a series. This will update the metadata of the series.
Args:
series_id (str): the ID of the series to analyze.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/analyze"
data = self.postRequest(url)
return data
def getSeriesBooks(
self,
series_id: str,
media_status: List[str] = None,
read_status: List[str] = None,
tag: List[str] = None,
deleted: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
sort: List[str] = None,
author: List[str] = None,
) -> List[Book]:
"""Get all books in a series.
Args:
series_id (str): The ID of the series to get books from.
media_status (List[str], optional): The status of the media. Can be: UNKNOWN,ERROR,READY,UNSUPPORTED,OUTDATED. Defaults to None.
read_status (List[str], optional): The read status of the book. Can be: UNREAD,READ,IN_PROGRESS. Defaults to None.
tag (List[str], optional): Tag(s) to query. Defaults to None.
deleted (bool, optional): If the series is deleted. 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.
page (int, optional): The page to be returned. Defaults to None.
size (int, optional): The size of the page. Defaults to None.
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.
Returns:
List[Book]: A list of all books that match the query. Each book is represented as a Book object.
"""
url = self.url + f"series/{series_id}/books"
params = locals()
params = self.setParams(params)
data = self.getRequest(url, params)
ret = []
for book in data["content"]:
ret.append(Book(**book))
return ret
def getSeriesCollections(self, series_id: str) -> Optional[List[Collection]]:
"""Get all collections assigned to a series.
Args:
series_id (str): the ID of the series to get collections from.
Returns:
Optional[List[Collection]]: A list of all collections that are assigned to the series. Each collection is represented as a Collection object. If no collections are assigned, None is returned.
"""
url = self.url + f"series/{series_id}/collections"
data = self.getRequest(url)
if data is None:
return None
ret = []
for collection in data:
ret.append(Collection(**collection))
return ret
def getFile(
self, series_id: str, filename: str, download_path: str = "~/Downloads"
) -> pathlib.Path:
"""Download the complete series as a giant zip file.
Unlike the other functions, this one has no timeout. It will wait until the file is downloaded.
WARNING: THIS WILL TAKE A VERY LONG TIME FOR LARGE SERIES. USE WITH CAUTION.
My test: 800mb took ~7min to download.
Args:
series_id (str): ID of the series to download the file from.
filename (str): The name of the file to download.
download_path (str, optional): The path to download the file to. Defaults to "~/Downloads".
Returns:
pathlib.Path: The path to the downloaded file.
"""
url = self.url + f"series/{series_id}/file"
download_path = pathlib.Path(download_path).expanduser()
subprocess.run(
[
"curl",
"-u",
f"{self._username}:{self._password}",
"-o",
f"{download_path}/{filename}",
url,
]
)
return pathlib.Path(f"{download_path}/{filename}")
@typing_extensions.deprecated("This function is not implemented yet.")
def deleteFile(self, series_id: str) -> None:
"""Not implemented yet."""
# url = self.url + f"series/{series_id}/file"
# data = self.deleteRequest(url)
raise NotImplementedError
def patchMetadata(self, series_id: str, changed_metadata: Dict[str, Any]) -> None:
"""Change the metadata of a series.
The changed metadata should be a dictionary with the changed fields. Only the fields that are changed need to be included.
Args:
series_id (str): The ID of the series to change.
changed_metadata (Dict[str,Any]): The changed metadata. This should be a dictionary with the changed fields. Only the fields that are changed need to be included.
Returns:
None
Example for full metadata:
{
"status": "ENDED",
"statusLock": true,
"title": "string",
"titleLock": true,
"titleSort": "string",
"titleSortLock": true,
"summary": "string",
"summaryLock": true,
"publisher": "string",
"publisherLock": true,
"readingDirectionLock": true,
"ageRatingLock": true,
"language": "string",
"languageLock": true,
"genresLock": true,
"tagsLock": true,
"totalBookCountLock": true,
"sharingLabelsLock": true,
"linksLock": true,
"alternateTitlesLock": true,
"tags": [
"string"
],
"links": [
{
"label": "string",
"url": "string"
}
],
"readingDirection": "LEFT_TO_RIGHT",
"ageRating": 0,
"genres": [
"string"
],
"totalBookCount": 0,
"sharingLabels": [
"string"
],
"alternateTitles": [
{
"label": "string",
"title": "string"
}
]
}
"""
url = self.url + f"series/{series_id}/metadata"
data = self.patchRequest(url, changed_metadata)
return data
def refreshMetadata(self, series_id: str) -> Optional[Dict[str, Any]]:
"""Refresh the metadata of a series.
Args:
series_id (str): The ID of the series to refresh.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/metadata/refresh"
data = self.postRequest(url)
return data
def setReadProgress(self, series_id: str) -> Optional[Dict[str, Any]]:
"""Mark all books in a series as read.
Args:
series_id (str): The ID of the series to get the read progress from.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/read-progress"
data = self.postRequest(url)
return data
def deleteReadProgress(self, series_id: str) -> Optional[Dict[str, Any]]:
"""Delete the read progress of a series.
Args:
series_id (str): The ID of the series to delete the read progress from.
Returns:
Optional[Dict[str,Any]] : If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/read-progress"
data = self.deleteRequest(url)
return data
def getSeriesThumbnail(self, series_id: str) -> str:
"""Get the current active thumbnail of a series.
Args:
series_id (str): the ID of the series to get the thumbnail from.
Returns:
str: the thumbnail of the series formatted as bytestring.
"""
url = self.url + f"series/{series_id}/thumbnail"
data = requests.get(url, auth=(self._username, self._password))
data = data.content
return data
def getSeriesThumbnails(self, series_id: str) -> List[Thumbnail]:
"""Get the metadata of all available thumbnails for the series.
Args:
series_id (str): the ID of the series to get the thumbnails from.
Returns:
List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class.
"""
url = self.url + f"series/{series_id}/thumbnails"
data = self.getRequest(url)
ret = []
for thumb in data:
ret.append(Thumbnail(**thumb))
return ret
def postThumbnail(
self, series_id: str, thumbnail: bytes, selected: bool = None
) -> Union[Thumbnail, Dict[str, Any]]:
"""Set a new thumbnail for the series.
Args:
thumbnail (bytes): The thumbnail to set. This should be a bytestring.
selected (bool, optional): If the thumbnail should be set as the active thumbnail. Defaults to None.
Returns:
None
"""
url = self.url + f"series/{series_id}/thumbnail"
data = self.postRequest(url, {"selected": selected, "thumbnail": thumbnail})
return Thumbnail(**data) if data else None
def getSpecificThumbnail(self, series_id: str, thumbnail_id: str) -> Thumbnail:
"""Get a specific thumbnail of a series.
Args:
series_id (str): The ID of the series to get the thumbnail from.
thumbnail_id (str): The ID of the thumbnail to get.
Returns:
Thumbnail: The thumbnail wrapped in a Thumbnail class.
"""
url = self.url + f"series/{series_id}/thumbnails/{thumbnail_id}"
data = self.getRequest(url)
return Thumbnail(**data)
def deleteSpecificThumbnail(
self, series_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
"""Delete a specific thumbnail of a series.
Args:
series_id (str): The ID of the series to delete the thumbnail from.
thumbnail_id (str): The ID of the thumbnail to delete.
Returns:
Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/thumbnails/{thumbnail_id}"
data = self.deleteRequest(url)
return data
def setThumbnail(
self, series_id: str, thumbnail_id: str
) -> Optional[Dict[str, Any]]:
"""Set a specific thumbnail as the active thumbnail of a series.
Args:
series_id (str): The ID of the series to set the thumbnail to.
thumbnail_id (str): The ID of the thumbnail to set.
Returns:
Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned.
"""
url = self.url + f"series/{series_id}/thumbnail/{thumbnail_id}/selected"
data = self.putRequest(url)
return data
def getAlphabeticalGroups(
self,
search_string: str = None,
library_id: List[str] = None,
collection_id: List[str] = None,
status: List[str] = None,
read_status: List[str] = None,
publisher: List[str] = None,
language: List[str] = None,
genre: List[str] = None,
tag: List[str] = None,
age_rating: List[str] = None,
release_year: List[str] = None,
sharing_label: List[str] = None,
deleted: bool = None,
complete: bool = None,
oneshot: bool = None,
search_regex: str = None,
author: List[str] = None,
) -> List[dict]:
"""Get a list of all series sorted by the starting letter of the series name.
Args:
search_string (str): The string to search for. Defaults to None.1
library_id (List[str], optional): The library to search in. Defaults to None.
collection_id (List[str], optional): The collection to search in. Defaults to None.
status (List[str], optional): The status of the series. Available values : ENDED, ONGOING, ABANDONED, HIATUS. Defaults to None.
read_status (List[str], optional): The read status of the series. Available values : UNREAD, READ, IN_PROGRESS. Defaults to None.
publisher (List[str], optional): The publisher(s) to search for. Defaults to None.
language (List[str], optional): The language(s) to search for. Defaults to None.
genre (List[str], optional): The genre(s) to search for. Defaults to None.
tag (List[str], optional): The tag(s) to search for. Defaults to None.
age_rating (List[str], optional): The age rating to filter for. Needs to be set manually. Defaults to None.
release_year (List[str], optional): The release year(s) to limit the search. Defaults to None.
sharing_label (List[str], optional): The sharing label(s) to . Defaults to None.
deleted (bool, optional): If deleted entries should be shown as well. Defaults to None.
complete (bool, optional): If only complete entries should be shown. Defaults to None.
oneshot (bool, optional): _description_. Defaults to None.
search_regex (str, optional): _description_. Defaults to None.
author (List[str], optional): _description_. Defaults to None.
Returns:
List[dict]: _description_
"""
params = locals()
params = self.setParams(params)
url = self.url + "series/alphabetical-groups"
data = self.getRequest(url, params)
return data
def getLatestSeries(
self,
library_id: List[str] = None,
deleted: bool = None,
oneshot: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
) -> List[Series]:
"""Get the latest changed series from the server.
Args:
library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None.
deleted (bool, optional): Set to true to include the deleted series. Defaults to None.
oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None.
unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True.
page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None.
size (int, optional): Defines the size of the page. Defaults to None.
Returns:
List[Series]: A list of all series that match the query. Each series is represented as a Series object.
"""
url = self.url + "series/latest"
params = locals()
params = self.setParams(params)
data = self.getRequest(url, params)
ret = []
for series in data["content"]:
ret.append(Series(**series))
return ret
def getNewSeries(
self,
library_id: List[str] = None,
deleted: bool = None,
oneshot: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
) -> List[Series]:
"""Get the newest series from the server.
Args:
library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None.
deleted (bool, optional): Set to true to include the deleted series. Defaults to None.
oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None.
unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True.
page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None.
size (int, optional): Defines the size of the page. Defaults to None.
Returns:
List[Series]: A list of all series that match the query. Each series is represented as a Series object.
"""
url = self.url + "series/new"
params = locals()
params = self.setParams(params)
data = self.getRequest(url, params)
ret = []
for series in data["content"]:
ret.append(Series(**series))
return ret
def getUpdatedSeries(
self,
library_id: List[str] = None,
deleted: bool = None,
oneshot: bool = None,
unpaged: bool = True,
page: int = None,
size: int = None,
) -> List["Series"]:
"""Get the latest updated series from the server.
Args:
library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None.
deleted (bool, optional): Set to true to include the deleted series. Defaults to None.
oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None.
unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True.
page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None.
size (int, optional): Defines the size of the page. Defaults to None.
Returns:
List[Series]: A list of all series that match the query. Each series is represented as a Series object.
"""
params = locals()
params = self.setParams(params)
url = self.url + "series/updated"
data = self.getRequest(url, params)
ret = []
for series in data["content"]:
ret.append(Series(**series))
return ret

22
src/komgapi/endpoints/settings_controller.py Normal file
View File

@@ -0,0 +1,22 @@
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
class SettingsController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None:
super().__init__(username, password, url, timeout)
def getSettings(self) -> Settings:
url = self.url + "settings"
data = self.getRequest(url)
return Settings(**data)
def updateSettings(self, settings: Settings) -> None:
url = self.url + "settings"
data = self.patchRequest(url, settings.model_dump())

92
src/komgapi/endpoints/user_controller.py Normal file
View File

@@ -0,0 +1,92 @@
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
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 getUsers(self) -> List[User]:
url = self.url + "users"
data = self.getRequest(url)
return [User(**user) for user in data]
def createUser(self, user: CreateUser) -> User:
url = self.url + "users"
data = self.postRequest(url, user.model_dump())
return User(**data)
def deleteUser(self, user_id: str) -> None:
url = self.url + f"users/{user_id}"
self.deleteRequest(url)
def updateUser(self, user_id: str, user: User) -> None:
userData = user.model_dump()
# remove id, email and password
userData.pop("id")
userData.pop("email")
userData.pop("password")
url = self.url + f"users/{user_id}"
data = self.patchRequest(url, user)
def getLatestAuthenticationActivity(self, user_id, api_key):
raise NotImplementedError
def changePassword(self, user_id: str, password: str) -> None:
url = self.url + f"users/{user_id}/password"
self.patchRequest(url, {"password": password})
def getUserAuthenticationActivity(
self,
unpaged: bool = None,
page: int = 0,
size: int = 20,
sort: List[str] = None,
) -> Authentication:
url = self.url + "users/authentification-activity"
params = self.setParams(locals())
data = self.getRequest(url, params)
return Authentication(**data)
def getMe(self) -> User | Violation:
url = self.url + "users/me"
data = self.getRequest(url)
return User(**data) if "id" in data else Violation(**data)
def getMyAPIKeys(self) -> List[APIKey] | Violation:
url = self.url + "users/me/api-keys"
data = self.getRequest(url)
return [APIKey(**data) for key in data] if "id" in data else []
def createAPIKey(self, key: str) -> APIKey:
url = self.url + "users/me/api-keys"
data = self.postRequest(url, f'"comment":{key}')
return APIKey(**data)
def deleteAPIKey(self, key_id: str) -> None:
url = self.url + f"users/me/api-keys/{key_id}"
self.deleteRequest(url)
def getMyAuthenticationActivity(
self,
unpaged: bool = None,
page: int = 0,
size: int = 20,
sort: List[str] = None,
) -> Authentication:
raise NotImplementedError
url = self.url + "users/me/authentication-activity"
params = self.setParams(locals())
data = self.getRequest(url, params)
print(data)
return UserAuthActivity(**data)
def changeMyPassword(self, password: str) -> None:
url = self.url + "users/me/password"
self.patchRequest(url, {"password": password})