Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 803de5b01d | |||
|
c55562a019
|
|||
|
1877905473
|
@@ -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 }}
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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"),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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]]:
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
10
src/komgapi/schemas/Error.py
Normal file
10
src/komgapi/schemas/Error.py
Normal 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
|
||||||
@@ -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=":/?&=;#@!$'()*+,;[]")
|
||||||
|
|||||||
@@ -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,
|
||||||
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user