From 1877905473728a0f553834d78b8d08b2ef7aba77 Mon Sep 17 00:00:00 2001 From: WorldTeacher Date: Tue, 15 Jul 2025 18:46:38 +0200 Subject: [PATCH 1/2] feat: Add retry logic and logging for POST request failures in BaseAPI --- src/komgapi/endpoints/baseapi.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/komgapi/endpoints/baseapi.py b/src/komgapi/endpoints/baseapi.py index 84f7007..3bdfc44 100644 --- a/src/komgapi/endpoints/baseapi.py +++ b/src/komgapi/endpoints/baseapi.py @@ -7,7 +7,7 @@ from limit import limit # type:ignore import loguru import sys import json - +import time log = loguru.logger log.remove() log.add("logs/komga_api.log", rotation="1 week", retention="1 month") @@ -96,18 +96,27 @@ class BaseAPI: json.dumps(data), json.dumps(body), ) - response.raise_for_status() status_code = response.status_code if status_code == 202: return None elif status_code == 200: return response.json() else: - raise ResultErrror(f"Result Error: {response.content}") + 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: - raise KomgaError(f"Timeout Error: {e}") from e + time.sleep(5) # Wait before retrying + log.error("Timeout during POST to {}: {}", url, str(e)) + + # raise KomgaError(f"Timeout Error: {e}") from e def patchRequest(self, url: str, data: Union[dict[Any, Any], None] = None): """Send PATCH request to API endpoint. From c55562a019e32fa7540e5828d26dde2c176137a5 Mon Sep 17 00:00:00 2001 From: WorldTeacher Date: Sat, 6 Dec 2025 15:37:31 +0100 Subject: [PATCH 2/2] add current files --- .gitea/workflows/release.yml | 133 +++++----- .../endpoints/announcement_controller.py | 28 ++- src/komgapi/endpoints/baseapi.py | 238 +++++++++++------- src/komgapi/endpoints/book_controller.py | 26 +- src/komgapi/endpoints/claim_controller.py | 28 ++- src/komgapi/endpoints/common_controller.py | 23 +- src/komgapi/endpoints/library_controller.py | 23 +- src/komgapi/endpoints/readlist_controller.py | 27 +- .../endpoints/referential_controller.py | 48 ++-- .../endpoints/series_collection_controller.py | 26 +- src/komgapi/endpoints/series_controller.py | 36 ++- src/komgapi/endpoints/settings_controller.py | 28 ++- src/komgapi/endpoints/user_controller.py | 30 ++- src/komgapi/komgapi.py | 112 ++++++--- src/komgapi/schemas/Error.py | 10 + src/komgapi/schemas/Link.py | 9 +- src/komgapi/schemas/Metadata.py | 19 ++ src/komgapi/schemas/__init__.py | 20 +- 18 files changed, 570 insertions(+), 294 deletions(-) create mode 100644 src/komgapi/schemas/Error.py diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml index a77bb37..82ac8dc 100644 --- a/.gitea/workflows/release.yml +++ b/.gitea/workflows/release.yml @@ -1,80 +1,83 @@ on: workflow_dispatch: inputs: - release_notes: - description: Release notes (use \n for newlines) - type: string - required: false github_release: - description: 'Create Gitea Release' + description: "Create Gitea Release" default: true type: boolean bump: - description: 'Bump type' + description: "Bump type" required: true - default: 'patch' + default: "patch" type: choice options: - - 'major' - - 'minor' - - 'patch' - + - "major" + - "minor" + - "patch" + jobs: build: runs-on: ubuntu-latest steps: - - name: Checkout code - uses: actions/checkout@master - - name: Install uv - uses: astral-sh/setup-uv@v5 - - name: Set up Python - run: uv python install - - name: Set Git identity - run: | - git config user.name "Gitea CI" - git config user.email "ci@git.theprivateserver.de" - - name: Bump version - id: bump - run: | - uv tool install bump-my-version - uv tool run bump-my-version bump ${{ github.event.inputs.bump }} - # echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version - echo "VERSION<> $GITHUB_ENV - echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV - echo "EOF" >> $GITHUB_ENV - - name: Push changes - uses: ad-m/github-push-action@master - with: - github_token: ${{ secrets.GITHUB_TOKEN }} - branch: ${{ github.ref }} - - - name: Create release notes - run: | - mkdir release_notes - echo -e "${{ inputs.release_notes }}" >> release_notes/release_notes.md - echo "Release notes:" - cat release_notes/release_notes.md - echo "" - - name: Build package - run: uv build - - name: Publish package - env: - USERNAME: ${{ github.repository_owner }} - run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }} - + - name: Checkout code + uses: actions/checkout@master + with: + fetch-depth: 0 + fetch-tags: true + - name: Install uv + uses: astral-sh/setup-uv@v5 + - name: Set up Python + run: uv python install + with: + python-version-file: "pyproject.toml" + - name: Set Git identity + run: | + git config user.name "Gitea CI" + git config user.email "ci@git.theprivateserver.de" + - name: Bump version + id: bump + run: | + uv tool install bump-my-version + uv tool run bump-my-version bump ${{ github.event.inputs.bump }} + # echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version + echo "VERSION<> $GITHUB_ENV + echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV + echo "EOF" >> $GITHUB_ENV + - name: Push changes + uses: ad-m/github-push-action@master + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + branch: ${{ github.ref }} + - name: Build Changelog + id: build_changelog + uses: https://github.com/mikepenz/release-changelog-builder-action@v5 + with: + platform: "gitea" + baseURL: "http://192.168.178.110:3000" + configuration: ".gitea/changelog_config.json" - - name: Create release - id: create_release - if: ${{ github.event.inputs.github_release == 'true' }} - uses: softprops/action-gh-release@master - with: - tag_name: ${{ env.VERSION }} - release_name: Release ${{ env.VERSION }} - body_path: release_notes/release_notes.md - draft: false - prerelease: false - make_latest: true - files: | - dist/* - env: - GITHUB_TOKEN: ${{ secrets.TOKEN }} \ No newline at end of file + env: + GITHUB_TOKEN: ${{ secrets.GITEA_TOKEN }} + - name: Build package + run: uv build + - name: Publish package + env: + USERNAME: ${{ github.repository_owner }} + run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }} + + - name: Create release + uses: softprops/action-gh-release@master + id: create_release + if: ${{ github.event.inputs.github_release == 'true' }} + with: + tag_name: v${{ env.VERSION }} + release_name: Release v${{ env.VERSION }} + body: ${{steps.build_changelog.outputs.changelog}} + draft: false + prerelease: false + make_latest: true + files: | + dist/* + env: + GITHUB_TOKEN: ${{ secrets.TOKEN }} + GITHUB_REPOSITORY: ${{ github.repository }} diff --git a/src/komgapi/endpoints/announcement_controller.py b/src/komgapi/endpoints/announcement_controller.py index ffa684c..48694fc 100644 --- a/src/komgapi/endpoints/announcement_controller.py +++ b/src/komgapi/endpoints/announcement_controller.py @@ -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" diff --git a/src/komgapi/endpoints/baseapi.py b/src/komgapi/endpoints/baseapi.py index 3bdfc44..2d59d69 100644 --- a/src/komgapi/endpoints/baseapi.py +++ b/src/komgapi/endpoints/baseapi.py @@ -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"), ) diff --git a/src/komgapi/endpoints/book_controller.py b/src/komgapi/endpoints/book_controller.py index c03589e..e5287b2 100644 --- a/src/komgapi/endpoints/book_controller.py +++ b/src/komgapi/endpoints/book_controller.py @@ -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( diff --git a/src/komgapi/endpoints/claim_controller.py b/src/komgapi/endpoints/claim_controller.py index e2a2360..2c96861 100644 --- a/src/komgapi/endpoints/claim_controller.py +++ b/src/komgapi/endpoints/claim_controller.py @@ -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" diff --git a/src/komgapi/endpoints/common_controller.py b/src/komgapi/endpoints/common_controller.py index 08474fb..51d6339 100644 --- a/src/komgapi/endpoints/common_controller.py +++ b/src/komgapi/endpoints/common_controller.py @@ -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" diff --git a/src/komgapi/endpoints/library_controller.py b/src/komgapi/endpoints/library_controller.py index 96f07a0..941e86c 100644 --- a/src/komgapi/endpoints/library_controller.py +++ b/src/komgapi/endpoints/library_controller.py @@ -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" diff --git a/src/komgapi/endpoints/readlist_controller.py b/src/komgapi/endpoints/readlist_controller.py index 4f05492..db367a1 100644 --- a/src/komgapi/endpoints/readlist_controller.py +++ b/src/komgapi/endpoints/readlist_controller.py @@ -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, diff --git a/src/komgapi/endpoints/referential_controller.py b/src/komgapi/endpoints/referential_controller.py index dc7e81d..9fd21e1 100644 --- a/src/komgapi/endpoints/referential_controller.py +++ b/src/komgapi/endpoints/referential_controller.py @@ -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() diff --git a/src/komgapi/endpoints/series_collection_controller.py b/src/komgapi/endpoints/series_collection_controller.py index f9745a4..06a3e23 100644 --- a/src/komgapi/endpoints/series_collection_controller.py +++ b/src/komgapi/endpoints/series_collection_controller.py @@ -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, diff --git a/src/komgapi/endpoints/series_controller.py b/src/komgapi/endpoints/series_controller.py index e3f6fd4..e1a6ed9 100644 --- a/src/komgapi/endpoints/series_controller.py +++ b/src/komgapi/endpoints/series_controller.py @@ -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]]: diff --git a/src/komgapi/endpoints/settings_controller.py b/src/komgapi/endpoints/settings_controller.py index 03ad4cd..d8ffdbd 100644 --- a/src/komgapi/endpoints/settings_controller.py +++ b/src/komgapi/endpoints/settings_controller.py @@ -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" diff --git a/src/komgapi/endpoints/user_controller.py b/src/komgapi/endpoints/user_controller.py index 40ad26c..651a2e5 100644 --- a/src/komgapi/endpoints/user_controller.py +++ b/src/komgapi/endpoints/user_controller.py @@ -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" diff --git a/src/komgapi/komgapi.py b/src/komgapi/komgapi.py index ecfae8b..ecf3f70 100644 --- a/src/komgapi/komgapi.py +++ b/src/komgapi/komgapi.py @@ -1,6 +1,7 @@ """Generic API for KOMGA""" -import requests +from typing import Optional + from .endpoints import * @@ -13,41 +14,106 @@ class KOMGAPI_REST: username (str): The username to use for the API. password (str): The password to use for the API. url (str): The URL of the KOMGA server. This should be the base URL of the server, without any paths. - timeout (int): The timeout for the requests. Defaults to 20 seconds. + timeout=timeout (int): The timeout=timeout for the requests. Defaults to 20 seconds. Example: ------- data= KOMGAPI_REST('username', 'password', 'http://localhost:8080/') """ - def __init__(self, username, password, url, timeout=20) -> None: + def __init__( + self, + username: Optional[str] = None, + password: Optional[str] = None, + api_key: Optional[str] = None, + url: str = "", + timeout: int = 20, + ) -> None: self._username = username self._password = password + self._api_key = api_key self.url = url - self.timeout = timeout + self.timeout = timeout = timeout = timeout if not url.endswith("/"): url += "/" self.common_book_controller = CommonBookController( - username, password, url, timeout + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.series_controller = SeriesController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.readlist_controller = ReadListController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, ) - self.series_controller = SeriesController(username, password, url, timeout) - self.readlist_controller = ReadListController(username, password, url, timeout) - self.page_hash_controller = None - self.library_controller = LibraryController(username, password, url, timeout) self.series_collection_controller = SeriesCollectionController( - username, password, url, timeout + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.book_controller = BookController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, ) - self.book_controler = BookController(username, password, url, timeout) self.announcement_controller = AnnouncementController( - username, password, url, timeout + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.user_controller = UserController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, ) - self.user_controller = UserController(username, password, url, timeout) self.transient_books_controller = None self.file_system_controller = None - self.claim_controller = ClaimController(username, password, url, timeout) - self.settings_controller = SettingsController(username, password, url, timeout) + self.claim_controller = ClaimController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.settings_controller = SettingsController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) self.referential_controller = ReferentialController( - username, password, url, timeout + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, + ) + self.library_controller = LibraryController( + username=username, + password=password, + api_key=api_key, + url=url, + timeout=timeout, ) self.login_controller = None self.historical_events_controller = None @@ -72,20 +138,6 @@ class KOMGAPI_REST: os.environ["KOMGA_URL"], ) - def test_connection(self): - """Test the connection to the KOMGA server. - - Returns: - bool: True if the connection is successful, False otherwise. - """ - try: - requests.get( - self.url, auth=(self._username, self._password), timeout=self.timeout - ) - return True - except Exception as e: - return False - # Book controller def seriesList(self) -> list[str]: """Get the list of books from the server. diff --git a/src/komgapi/schemas/Error.py b/src/komgapi/schemas/Error.py new file mode 100644 index 0000000..f299af9 --- /dev/null +++ b/src/komgapi/schemas/Error.py @@ -0,0 +1,10 @@ +from dataclasses import dataclass + + +@dataclass +class Error: + timestamp: str = None + status: int = None + error: str = None + message: str = None + path: str = None diff --git a/src/komgapi/schemas/Link.py b/src/komgapi/schemas/Link.py index 5812abb..8a4790a 100644 --- a/src/komgapi/schemas/Link.py +++ b/src/komgapi/schemas/Link.py @@ -1,11 +1,16 @@ from __future__ import annotations -from typing import List - from dataclasses import dataclass +from urllib.parse import quote @dataclass class Link: label: str url: str + + def __post_init__(self): + #set url to use unicode characters + url_prefix = self.url.split("://")[0] + "://" + url_content = self.url.split("://")[1] + self.url = url_prefix + quote(url_content, safe=":/?&=;#@!$'()*+,;[]") diff --git a/src/komgapi/schemas/Metadata.py b/src/komgapi/schemas/Metadata.py index 89727eb..621ac1d 100644 --- a/src/komgapi/schemas/Metadata.py +++ b/src/komgapi/schemas/Metadata.py @@ -43,3 +43,22 @@ class Metadata: alternateTitlesLock: bool = None created: Optional[str] | None = None lastModified: Optional[str] | None = None + + + + @property + def print(self)->dict[str, Optional[Union[str, int, bool, List[str], List[str], List[str]]]]: + return { + "status": self.status, + "title": self.title, + "titleSort": self.titleSort, + "summary": self.summary, + "publisher": self.publisher, + "ageRating": self.ageRating, + "language": self.language, + "genres": ", ".join(self.genres) if self.genres else None, + "tags": ", ".join(self.tags) if self.tags else None, + "totalBookCount": self.totalBookCount, + "links": [link["url"] for link in self.links] if self.links else None, + "alternateTitles": [alternate["title"] for alternate in self.alternateTitles] if self.alternateTitles else None, + } diff --git a/src/komgapi/schemas/__init__.py b/src/komgapi/schemas/__init__.py index 6c8ad7a..1f57ef6 100644 --- a/src/komgapi/schemas/__init__.py +++ b/src/komgapi/schemas/__init__.py @@ -1,11 +1,15 @@ from .AlternateTitle import AlternateTitle +from .Announcement import Announcement +from .apikey import APIKey +from .Authentication import Authentication, UserAuthActivity from .Author import Author, AuthorResponse from .Book import Book -from .BooksMetadata import BooksMetadata, BookMetadata +from .BooksMetadata import BookMetadata, BooksMetadata from .Collection import Collection from .Duplicate import Duplicate +from .Error import Error from .Latest import LatestSeriesData -from .Library import Library, CreateLibrary +from .Library import CreateLibrary, Library from .Link import Link from .Locations import Locations from .Manifest import Manifest @@ -17,14 +21,10 @@ from .Position import Position from .Progress import Progress from .Readlist import Readlist from .Series import Series -from .Sort import Sort -from .Text import Text -from .Thumbnail import Thumbnail, ReadlistThumbnail -from .Announcement import Announcement -from .User import User, CreateUser -from .Authentication import Authentication, UserAuthActivity -from .apikey import APIKey from .settings import Settings +from .Sort import Sort from .status import Status - +from .Text import Text +from .Thumbnail import ReadlistThumbnail, Thumbnail +from .User import CreateUser, User from .violation import Violation