3 Commits
0.1.1 ... main

18 changed files with 568 additions and 283 deletions

View File

@@ -1,80 +1,83 @@
on: on:
workflow_dispatch: workflow_dispatch:
inputs: inputs:
release_notes:
description: Release notes (use \n for newlines)
type: string
required: false
github_release: github_release:
description: 'Create Gitea Release' description: "Create Gitea Release"
default: true default: true
type: boolean type: boolean
bump: bump:
description: 'Bump type' description: "Bump type"
required: true required: true
default: 'patch' default: "patch"
type: choice type: choice
options: options:
- 'major' - "major"
- 'minor' - "minor"
- 'patch' - "patch"
jobs: jobs:
build: build:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: Checkout code - name: Checkout code
uses: actions/checkout@master uses: actions/checkout@master
- name: Install uv with:
uses: astral-sh/setup-uv@v5 fetch-depth: 0
- name: Set up Python fetch-tags: true
run: uv python install - name: Install uv
- name: Set Git identity uses: astral-sh/setup-uv@v5
run: | - name: Set up Python
git config user.name "Gitea CI" run: uv python install
git config user.email "ci@git.theprivateserver.de" with:
- name: Bump version python-version-file: "pyproject.toml"
id: bump - name: Set Git identity
run: | run: |
uv tool install bump-my-version git config user.name "Gitea CI"
uv tool run bump-my-version bump ${{ github.event.inputs.bump }} git config user.email "ci@git.theprivateserver.de"
# echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version - name: Bump version
echo "VERSION<<EOF" >> $GITHUB_ENV id: bump
echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV run: |
echo "EOF" >> $GITHUB_ENV uv tool install bump-my-version
- name: Push changes uv tool run bump-my-version bump ${{ github.event.inputs.bump }}
uses: ad-m/github-push-action@master # echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version
with: echo "VERSION<<EOF" >> $GITHUB_ENV
github_token: ${{ secrets.GITHUB_TOKEN }} echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV
branch: ${{ github.ref }} echo "EOF" >> $GITHUB_ENV
- name: Push changes
- name: Create release notes uses: ad-m/github-push-action@master
run: | with:
mkdir release_notes github_token: ${{ secrets.GITHUB_TOKEN }}
echo -e "${{ inputs.release_notes }}" >> release_notes/release_notes.md branch: ${{ github.ref }}
echo "Release notes:" - name: Build Changelog
cat release_notes/release_notes.md id: build_changelog
echo "" uses: https://github.com/mikepenz/release-changelog-builder-action@v5
- name: Build package with:
run: uv build platform: "gitea"
- name: Publish package baseURL: "http://192.168.178.110:3000"
env: configuration: ".gitea/changelog_config.json"
USERNAME: ${{ github.repository_owner }}
run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }}
- name: Create release env:
id: create_release GITHUB_TOKEN: ${{ secrets.GITEA_TOKEN }}
if: ${{ github.event.inputs.github_release == 'true' }} - name: Build package
uses: softprops/action-gh-release@master run: uv build
with: - name: Publish package
tag_name: ${{ env.VERSION }} env:
release_name: Release ${{ env.VERSION }} USERNAME: ${{ github.repository_owner }}
body_path: release_notes/release_notes.md run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }}
draft: false
prerelease: false - name: Create release
make_latest: true uses: softprops/action-gh-release@master
files: | id: create_release
dist/* if: ${{ github.event.inputs.github_release == 'true' }}
env: with:
GITHUB_TOKEN: ${{ secrets.TOKEN }} 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 }}

View File

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

View File

@@ -1,12 +1,12 @@
import sys
from typing import Any, Optional, Union
import httpx import httpx
import loguru
from httpx_retries import Retry, RetryTransport from httpx_retries import Retry, RetryTransport
from komgapi.errors import KomgaError, ResultErrror
from typing import Any, Union
from limit import limit # type:ignore from limit import limit # type:ignore
import loguru from komgapi.errors import KomgaError
import sys
import json
log = loguru.logger log = loguru.logger
log.remove() log.remove()
@@ -17,22 +17,67 @@ log.add(sys.stdout, level="INFO")
class BaseAPI: class BaseAPI:
def __init__( def __init__(
self, self,
username: str, username: Optional[str] = None,
password: str, password: Optional[str] = None,
url: str, api_key: Optional[str] = None,
url: str = "",
timeout: int = 20, timeout: int = 20,
api_version: int = 1, api_version: int = 1,
) -> None: ) -> 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.url = url + f"api/v{api_version}/"
self.timeout = timeout self.timeout = timeout
self.headers = { self.api_key = api_key
"Content-Type": "application/json", self.username = username
"Accept": "application/json", 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]: 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 { return {
param_name: param param_name: param
for param_name, param in locals.items() for param_name, param in locals.items()
@@ -41,36 +86,58 @@ class BaseAPI:
} }
def test_connection(self): def test_connection(self):
"""
Test the connection to the API.
Returns:
bool: True if the connection is successful, False otherwise.
"""
try: try:
with httpx.Client(timeout=self.timeout) as client: with httpx.Client(timeout=self.timeout, headers=self.headers) as client:
client.get(self.url, headers=self.headers) client.get(self.url, auth=self._get_auth())
return True return True
except httpx.RequestError: except httpx.RequestError:
return False return False
def overwriteVersion(self, version: int): 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}/") self.url = self.url.replace("api/v1/", f"api/v{version}/")
return self return self
@limit(1, 1) @limit(1, 1)
def getRequest(self, url: str, params: Union[dict[Any, Any], None] = None) -> Any: 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: if params is None:
params = {} params = {}
try: try:
with httpx.Client( with httpx.Client(
timeout=self.timeout, timeout=self.timeout,
auth=(self._username, self._password), auth=self._get_auth(),
transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5)), transport=RetryTransport(retry=Retry(total=5, backoff_factor=0.5)),
headers=self.headers,
) as client: ) as client:
response = client.get(url, params=params, headers=self.headers) response = client.get(url, params=params)
if response.status_code != 200:
return self.getRequest(url, params)
return response.json() return response.json()
except httpx.ConnectError as e: except httpx.RequestError as e:
raise KomgaError(f"Connection Error: {e}") from e raise KomgaError(f"Request Error: {e}") from e
except httpx.TimeoutException as e:
raise KomgaError(f"Timeout Error: {e}") from e
def postRequest( def postRequest(
self, self,
@@ -78,102 +145,96 @@ class BaseAPI:
data: Union[dict[Any, Any], None] = None, data: Union[dict[Any, Any], None] = None,
body: 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: if data is None:
data = {} data = {}
try: try:
with httpx.Client( 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: ) as client:
response = client.post( response = client.post(
url, url,
params=data, params=data,
json=body if body is not None else {}, json=body if body is not None else {},
headers=self.headers,
) )
log.debug(
"POST request to {} with data: {}, json: {}",
url,
json.dumps(data),
json.dumps(body),
)
response.raise_for_status() response.raise_for_status()
status_code = response.status_code return response.json() if response.content else None
if status_code == 202: except httpx.RequestError as e:
return None raise KomgaError(f"Request Error: {e}") from e
elif status_code == 200:
return response.json()
else:
raise ResultErrror(f"Result Error: {response.content}")
except httpx.ConnectError as e:
raise KomgaError(f"Connection Error: {e}") from e
except httpx.TimeoutException as e:
raise KomgaError(f"Timeout Error: {e}") from e
def patchRequest(self, url: str, data: Union[dict[Any, Any], None] = None): 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: Args:
url (str): API endpoint URL url (str): API endpoint URL.
data (Union[dict[Any, Any], None]): Data to send in request body data (Union[dict[Any, Any], None]): Request body.
Returns: Returns:
dict: JSON response from server Any: JSON response from the server.
Raises:
KomgaError: For connection/timeout errors
ResultError: For invalid responses
""" """
if data is None: if data is None:
data = {} data = {}
try: try:
log.debug("PATCH request to {} with data: {}", url, json.dumps(data))
with httpx.Client( with httpx.Client(
timeout=self.timeout, timeout=self.timeout,
auth=(self._username, self._password), auth=self._get_auth(),
headers=self.headers, headers=self.headers,
) as client: ) as client:
response = client.patch(url, json=data) response = client.patch(url, json=data)
response.raise_for_status() response.raise_for_status()
if response.status_code == 204:
return None
return response.json() if response.content else None return response.json() if response.content else None
except httpx.RequestError as e:
except httpx.ConnectError as e: raise KomgaError(f"Request Error: {e}") from 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
def deleteRequest(self, url: str): 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: try:
with httpx.Client( with httpx.Client(
timeout=self.timeout, auth=(self._username, self._password) timeout=self.timeout,
auth=self._get_auth(),
headers=self.headers,
) as client: ) as client:
response = client.delete(url, headers=self.headers) response = client.delete(url)
response.raise_for_status() response.raise_for_status()
return response.json() return response.json() if response.content else None
except httpx.ConnectError as e: except httpx.RequestError as e:
raise KomgaError(f"Connection Error: {e}") from e raise KomgaError(f"Request Error: {e}") from e
except httpx.TimeoutException as e:
raise KomgaError(f"Timeout Error: {e}") from e
@classmethod @classmethod
def from_env(cls): def from_env(cls):
"""
Create a BaseAPI instance using environment variables.
Returns:
BaseAPI: Configured BaseAPI instance.
"""
import os import os
return cls( return cls(
os.environ["KOMGA_USERNAME"], url=os.environ["KOMGA_URL"],
os.environ["KOMGA_PASSWORD"], username=os.environ.get("KOMGA_USERNAME"),
os.environ["KOMGA_URL"], password=os.environ.get("KOMGA_PASSWORD"),
api_key=os.environ.get("KOMGA_API_KEY"),
) )

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -1,16 +1,28 @@
from .baseapi import BaseAPI from typing import Any, Dict, List, Optional
import pathlib
import subprocess
import requests import requests
import typing_extensions
from typing import List, Optional, Dict, Any
from komgapi.schemas import * # Progress, Series from komgapi.schemas import * # Progress, Series
from .baseapi import BaseAPI
class SeriesCollectionController(BaseAPI): class SeriesCollectionController(BaseAPI):
def __init__(self, username, password, url, timeout=20) -> None: def __init__(
super().__init__(username, password, url, timeout) 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( def getCollections(
self, self,

View File

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

View File

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

View File

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

View File

@@ -1,6 +1,7 @@
"""Generic API for KOMGA""" """Generic API for KOMGA"""
import requests from typing import Optional
from .endpoints import * from .endpoints import *
@@ -13,41 +14,106 @@ class KOMGAPI_REST:
username (str): The username to use for the API. username (str): The username to use for the API.
password (str): The password 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. 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: Example:
------- -------
data= KOMGAPI_REST('username', 'password', 'http://localhost:8080/') 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._username = username
self._password = password self._password = password
self._api_key = api_key
self.url = url self.url = url
self.timeout = timeout self.timeout = timeout = timeout = timeout
if not url.endswith("/"): if not url.endswith("/"):
url += "/" url += "/"
self.common_book_controller = CommonBookController( 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( 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( 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.transient_books_controller = None
self.file_system_controller = None self.file_system_controller = None
self.claim_controller = ClaimController(username, password, url, timeout) self.claim_controller = ClaimController(
self.settings_controller = SettingsController(username, password, url, timeout) 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( 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.login_controller = None
self.historical_events_controller = None self.historical_events_controller = None
@@ -72,20 +138,6 @@ class KOMGAPI_REST:
os.environ["KOMGA_URL"], 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 # Book controller
def seriesList(self) -> list[str]: def seriesList(self) -> list[str]:
"""Get the list of books from the server. """Get the list of books from the server.

View File

@@ -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

View File

@@ -1,11 +1,16 @@
from __future__ import annotations from __future__ import annotations
from typing import List
from dataclasses import dataclass from dataclasses import dataclass
from urllib.parse import quote
@dataclass @dataclass
class Link: class Link:
label: str label: str
url: 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=":/?&=;#@!$'()*+,;[]")

View File

@@ -43,3 +43,22 @@ class Metadata:
alternateTitlesLock: bool = None alternateTitlesLock: bool = None
created: Optional[str] | None = None created: Optional[str] | None = None
lastModified: 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,
}

View File

@@ -1,11 +1,15 @@
from .AlternateTitle import AlternateTitle from .AlternateTitle import AlternateTitle
from .Announcement import Announcement
from .apikey import APIKey
from .Authentication import Authentication, UserAuthActivity
from .Author import Author, AuthorResponse from .Author import Author, AuthorResponse
from .Book import Book from .Book import Book
from .BooksMetadata import BooksMetadata, BookMetadata from .BooksMetadata import BookMetadata, BooksMetadata
from .Collection import Collection from .Collection import Collection
from .Duplicate import Duplicate from .Duplicate import Duplicate
from .Error import Error
from .Latest import LatestSeriesData from .Latest import LatestSeriesData
from .Library import Library, CreateLibrary from .Library import CreateLibrary, Library
from .Link import Link from .Link import Link
from .Locations import Locations from .Locations import Locations
from .Manifest import Manifest from .Manifest import Manifest
@@ -17,14 +21,10 @@ from .Position import Position
from .Progress import Progress from .Progress import Progress
from .Readlist import Readlist from .Readlist import Readlist
from .Series import Series 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 .settings import Settings
from .Sort import Sort
from .status import Status from .status import Status
from .Text import Text
from .Thumbnail import ReadlistThumbnail, Thumbnail
from .User import CreateUser, User
from .violation import Violation from .violation import Violation