refactor: rework configuration models, add CLI, tests and VSCode settings
- Replace hardcoded CONFIG_PATH with appdirs AppDirs (use app.user_config_dir) and update SETTINGS_PATH - Rework dataclasses: add Library, QbitTorrent; convert Komga.libraries to List[Library]; use Path for media_path; add getLibraryByName and proper __post_init__ conversions - Extend KomGrabber to support downloader/downloader_settings (aria2/qbit), normalize Path handling and expanduser usage - Add type fixes, utility methods (getattr/_setattr) and API __post_init__ to convert nested dicts to dataclass instances - Add package CLI entrypoints (__main__.py, src package main) and simple runner (main.py) - Add tests for package __init__ CLI and config behavior (tests/*) and a small test script (test.py) - Add .vscode/settings.json for pytest integration - Tidy pyproject.toml: format dependencies, add dev/test dependency groups and fix trailing newline in bumpversion section
This commit is contained in:
7
.vscode/settings.json
vendored
Normal file
7
.vscode/settings.json
vendored
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"python.testing.pytestArgs": [
|
||||
"tests"
|
||||
],
|
||||
"python.testing.unittestEnabled": false,
|
||||
"python.testing.pytestEnabled": true
|
||||
}
|
||||
30
__main__.py
Normal file
30
__main__.py
Normal file
@@ -0,0 +1,30 @@
|
||||
import argparse
|
||||
from src.komconfig import KomConfig
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Retrieve configuration values.")
|
||||
parser.add_argument("key", type=str, help="The configuration key to retrieve.")
|
||||
args = parser.parse_args()
|
||||
|
||||
config = KomConfig()
|
||||
|
||||
try:
|
||||
config_dict = config.dict()
|
||||
if isinstance(config_dict, dict):
|
||||
value = config_dict.get(args.key, None)
|
||||
if value is None:
|
||||
print(f"Key '{args.key}' not found in the configuration.")
|
||||
else:
|
||||
try:
|
||||
print(str(value))
|
||||
except Exception:
|
||||
print(repr(value))
|
||||
else:
|
||||
print("Configuration data is not in dictionary format.")
|
||||
except Exception as e:
|
||||
print(f"Error retrieving key '{args.key}': {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
9
main.py
Normal file
9
main.py
Normal file
@@ -0,0 +1,9 @@
|
||||
from src.komconfig import KomConfig
|
||||
|
||||
|
||||
def test_komconfig():
|
||||
print(KomConfig().general.log_file)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_komconfig()
|
||||
@@ -5,14 +5,23 @@ description = "A small library providing a config class that provides settings d
|
||||
readme = "README.md"
|
||||
authors = [{ name = "WorldTeacher", email = "coding_contact@pm.me" }]
|
||||
requires-python = ">=3.13"
|
||||
dependencies = ["omegaconf>=2.3.0"]
|
||||
dependencies = [
|
||||
"appdirs>=1.4.4",
|
||||
"omegaconf>=2.3.0",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[dependency-groups]
|
||||
test = ["pytest>=8.3.4"]
|
||||
dev = [
|
||||
"pip>=25.1.1",
|
||||
]
|
||||
test = [
|
||||
"pytest>=8.3.4",
|
||||
"pytest-cov>=6.1.1",
|
||||
]
|
||||
|
||||
[tool.bumpversion]
|
||||
current_version = "0.2.0"
|
||||
@@ -34,4 +43,4 @@ moveable_tags = []
|
||||
commit_args = ""
|
||||
setup_hooks = []
|
||||
pre_commit_hooks = []
|
||||
post_commit_hooks = []
|
||||
post_commit_hooks = []
|
||||
|
||||
@@ -1,2 +1,38 @@
|
||||
CONFIG_PATH = "~/.config/KomSuite/"
|
||||
__all__ = ["KomConfig", "app"]
|
||||
|
||||
from appdirs import AppDirs
|
||||
|
||||
app = AppDirs("KomSuite", "KomConfig")
|
||||
|
||||
|
||||
import argparse
|
||||
|
||||
from .config import Settings as KomConfig
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Retrieve configuration values.")
|
||||
parser.add_argument("key", type=str, help="The configuration key to retrieve.")
|
||||
args = parser.parse_args()
|
||||
|
||||
config = KomConfig()
|
||||
|
||||
try:
|
||||
config_dict = config.dict()
|
||||
if isinstance(config_dict, dict):
|
||||
value = config_dict.get(args.key, None)
|
||||
if value is None:
|
||||
print(f"Key '{args.key}' not found in the configuration.")
|
||||
else:
|
||||
try:
|
||||
print(str(value))
|
||||
except Exception:
|
||||
print(repr(value))
|
||||
else:
|
||||
print("Configuration data is not in dictionary format.")
|
||||
except Exception as e:
|
||||
print(f"Error retrieving key '{args.key}': {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Union
|
||||
import os
|
||||
from omegaconf import OmegaConf, DictConfig
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from komconfig import CONFIG_PATH
|
||||
from typing import List, Optional, Union
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
SETTINGS_PATH = os.path.join(CONFIG_PATH, "config.yaml")
|
||||
from omegaconf import DictConfig, OmegaConf
|
||||
|
||||
from komconfig import app
|
||||
|
||||
SETTINGS_PATH = os.path.join(app.user_config_dir, "config.yaml")
|
||||
|
||||
|
||||
@dataclass
|
||||
class RemoteSettings:
|
||||
url: str
|
||||
@@ -19,11 +23,9 @@ class RemoteSettings:
|
||||
@dataclass
|
||||
class Cache:
|
||||
mode: str
|
||||
remote: dict[str, Union[str, int]] |RemoteSettings
|
||||
remote: dict[str, Union[str, int]] | RemoteSettings
|
||||
local_path: str | Path
|
||||
|
||||
|
||||
|
||||
def __post__init__(self):
|
||||
if self.mode == "remote":
|
||||
self.remote = RemoteSettings(**self.remote)
|
||||
@@ -37,12 +39,33 @@ class Cache:
|
||||
return Path(self.local_path).expanduser()
|
||||
else:
|
||||
return f"{self.remote.url}:{self.remote.port}"
|
||||
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
password = quote_plus(self.remote.password) if self.remote else ""
|
||||
return f"mysql+pymysql://{self.remote.user}:{password}@{self.remote.url}:{self.remote.port}/{self.remote.database_name}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Library:
|
||||
"""Komga library settings."""
|
||||
|
||||
id: Optional[str] = None
|
||||
name: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
media_path: Optional[str] = None
|
||||
valid_extensions: Optional[List[str]] = None
|
||||
|
||||
def from_ddict(self, d: dict):
|
||||
keys = d.keys()
|
||||
for key in keys:
|
||||
self.name = key
|
||||
data = d[self.name]
|
||||
for key, value in data.items():
|
||||
setattr(self, key, value)
|
||||
return self
|
||||
|
||||
|
||||
@dataclass
|
||||
class Komga:
|
||||
"""Komga server settings."""
|
||||
@@ -50,9 +73,9 @@ class Komga:
|
||||
url: str
|
||||
user: str
|
||||
password: str
|
||||
media_path: str
|
||||
media_path: Path
|
||||
api_key: str = None
|
||||
libraries: dict[str, str] = None
|
||||
libraries: List[Library] = None
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
@@ -62,7 +85,24 @@ class Komga:
|
||||
|
||||
def __post_init__(self):
|
||||
if "~" in self.media_path:
|
||||
self.media_path = os.path.expanduser(self.media_path)
|
||||
self.media_path = Path(os.path.expanduser(self.media_path))
|
||||
if self.libraries:
|
||||
self.libraries = [Library().from_ddict(lib) for lib in self.libraries]
|
||||
|
||||
def getLibraryByName(self, name: str) -> Optional[Library]:
|
||||
"""Get a library by its name.
|
||||
|
||||
Args:
|
||||
name (str): The name of the library.
|
||||
|
||||
Returns:
|
||||
Optional[Library]: The library object if found, None otherwise.
|
||||
"""
|
||||
if self.libraries:
|
||||
for lib in self.libraries:
|
||||
if lib.name == name:
|
||||
return lib
|
||||
return None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -96,6 +136,23 @@ class Aria2:
|
||||
setattr(self, name, value)
|
||||
|
||||
|
||||
@dataclass
|
||||
class QbitTorrent:
|
||||
"""QbitTorrent settings."""
|
||||
|
||||
host: str
|
||||
port: int
|
||||
username: str
|
||||
password: str
|
||||
category: str
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
|
||||
def _setattr(self, name, value):
|
||||
setattr(self, name, value)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EbookSettings:
|
||||
min_filesize: int
|
||||
@@ -168,22 +225,35 @@ class KomGrabber:
|
||||
"""
|
||||
|
||||
download_location: Path
|
||||
tag_location: Path
|
||||
copy_location: Optional[Path]
|
||||
copy_files: Optional[List[str]]
|
||||
copy: bool
|
||||
|
||||
get_chapters: bool
|
||||
skip_parameters: List[str]
|
||||
aria2: Aria2
|
||||
tag_interactive: bool
|
||||
ebook: EbookSettings
|
||||
manga: ComicSettings
|
||||
check_interval: int
|
||||
use_cache: bool
|
||||
cache_check_interval: int
|
||||
downloader: str
|
||||
downloader_settings: Union[Aria2, QbitTorrent]
|
||||
|
||||
def __post_init__(self):
|
||||
self.skip_parameters = [param.lower() for param in self.skip_parameters]
|
||||
if "~" in self.download_location:
|
||||
self.download_location = os.path.expanduser(self.download_location)
|
||||
# if "~" in self.download_location:
|
||||
# self.download_location = os.path.expanduser(self.download_location)
|
||||
if isinstance(self.download_location, str):
|
||||
self.download_location = Path(self.download_location)
|
||||
self.download_location = Path(self.download_location).expanduser()
|
||||
if isinstance(self.tag_location, str):
|
||||
self.tag_location = Path(self.tag_location).expanduser()
|
||||
if isinstance(self.downloader_settings, dict):
|
||||
if self.downloader == "aria2":
|
||||
self.downloader_settings = Aria2(**self.downloader_settings)
|
||||
elif self.downloader == "qbit":
|
||||
self.downloader_settings = QbitTorrent(**self.downloader_settings)
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
@@ -196,9 +266,9 @@ class KomGrabber:
|
||||
class KomTagger:
|
||||
"""KomTagger settings."""
|
||||
|
||||
failed_location: str
|
||||
success_location: str
|
||||
sanitize_description: bool
|
||||
check_cache: bool
|
||||
check_interval: int
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
@@ -207,18 +277,15 @@ class KomTagger:
|
||||
setattr(self, name, value)
|
||||
|
||||
def __post_init__(self):
|
||||
if "~" in self.failed_location:
|
||||
self.failed_location = os.path.expanduser(self.failed_location)
|
||||
if "~" in self.success_location:
|
||||
self.success_location = os.path.expanduser(self.success_location)
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class ComicVine:
|
||||
"""ComicVine settings."""
|
||||
|
||||
api_key: str
|
||||
url: str
|
||||
api_key: str
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
@@ -267,17 +334,11 @@ class API:
|
||||
mangadex: MangaDex
|
||||
comicsorg: ComicsOrg
|
||||
|
||||
@property
|
||||
def comicvine(self):
|
||||
return self.comicvine
|
||||
|
||||
@property
|
||||
def mangadex(self):
|
||||
return self.mangadex
|
||||
|
||||
@property
|
||||
def comicsorg(self):
|
||||
return self.comicsorg
|
||||
def __post_init__(self):
|
||||
# Convert dictionaries to their respective dataclass objects
|
||||
self.comicvine = ComicVine(**self.comicvine)
|
||||
self.mangadex = MangaDex(**self.mangadex)
|
||||
self.comicsorg = ComicsOrg(**self.comicsorg)
|
||||
|
||||
def getattr(self, name):
|
||||
return getattr(self, name)
|
||||
@@ -363,7 +424,7 @@ class Settings:
|
||||
@property
|
||||
def komtagger(self):
|
||||
return KomTagger(**self._config.komtagger)
|
||||
|
||||
|
||||
@property
|
||||
def cache(self):
|
||||
return Cache(**self._config.cache)
|
||||
|
||||
6
test.py
Normal file
6
test.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from src.komconfig import KomConfig
|
||||
|
||||
cfg = KomConfig()
|
||||
|
||||
|
||||
print(cfg.komga.libraries)
|
||||
40
tests/test___init__.py
Normal file
40
tests/test___init__.py
Normal file
@@ -0,0 +1,40 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from src.komconfig.__init__ import main
|
||||
import sys
|
||||
|
||||
|
||||
def test_valid_key():
|
||||
mock_config = MagicMock()
|
||||
mock_config.dict.return_value = {"valid_key": "value"}
|
||||
|
||||
with patch("src.komconfig.__init__.KomConfig", return_value=mock_config):
|
||||
with patch("sys.argv", ["__init__.py", "valid_key"]):
|
||||
main()
|
||||
|
||||
|
||||
def test_key_not_found():
|
||||
mock_config = MagicMock()
|
||||
mock_config.dict.return_value = {"another_key": "value"}
|
||||
|
||||
with patch("src.komconfig.__init__.KomConfig", return_value=mock_config):
|
||||
with patch("sys.argv", ["__init__.py", "missing_key"]):
|
||||
main()
|
||||
|
||||
|
||||
def test_non_dict_config():
|
||||
mock_config = MagicMock()
|
||||
mock_config.dict.return_value = None
|
||||
|
||||
with patch("src.komconfig.__init__.KomConfig", return_value=mock_config):
|
||||
with patch("sys.argv", ["__init__.py", "any_key"]):
|
||||
main()
|
||||
|
||||
|
||||
def test_exception_handling():
|
||||
mock_config = MagicMock()
|
||||
mock_config.dict.side_effect = Exception("Test Exception")
|
||||
|
||||
with patch("src.komconfig.__init__.KomConfig", return_value=mock_config):
|
||||
with patch("sys.argv", ["__init__.py", "any_key"]):
|
||||
main()
|
||||
18
tests/test_config.py
Normal file
18
tests/test_config.py
Normal file
@@ -0,0 +1,18 @@
|
||||
from komconfig import KomConfig
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def komconfig():
|
||||
return KomConfig()
|
||||
|
||||
|
||||
def test_komconfig(komconfig):
|
||||
assert komconfig is not None
|
||||
|
||||
|
||||
def test_general(komconfig):
|
||||
assert KomConfig().general.log_file == "/var/log/komgrabber.log"
|
||||
assert KomConfig().general.log_level == "INFO"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user