Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 803de5b01d | |||
|
c55562a019
|
|||
|
1877905473
|
@@ -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<<EOF" >> $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: 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<<EOF" >> $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 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 }}
|
||||
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
|
||||
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 }}
|
||||
- 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 }}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
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
|
||||
from komgapi.errors import KomgaError
|
||||
|
||||
log = loguru.logger
|
||||
log.remove()
|
||||
@@ -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,102 +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),
|
||||
)
|
||||
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}")
|
||||
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
|
||||
|
||||
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"),
|
||||
)
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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]]:
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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.
|
||||
|
||||
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 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=":/?&=;#@!$'()*+,;[]")
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user