[+] cache settings singleton, remove per-subcommand --cache-dir
1. add apps/cache/settings.py: cache_settings_t pydantic-settings singleton
with default dir ~/.cache/online.fxreader.pr34.commands_typed.archlinux/
apps/cache/cache_dir, env var ARCHLINUX_CACHE_DIR;
2. add apps/cache/cli.py: cache_cli_t with add_arguments/extract/apply;
3. remove --cache-dir from compile and archive subcommands, they now
use cache_settings_t.singleton().dir;
4. add test_cache_settings.py;
This commit is contained in:
parent
8079aae41c
commit
cd170d2e9e
36
python/online/fxreader/pr34/commands_typed/archlinux/apps/cache/cli.py
vendored
Normal file
36
python/online/fxreader/pr34/commands_typed/archlinux/apps/cache/cli.py
vendored
Normal file
@ -0,0 +1,36 @@
|
||||
"""CLI integration for cache settings.
|
||||
|
||||
Provides methods to inject argparse arguments, extract parsed values,
|
||||
and apply them to the cache settings singleton.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .settings import cache_settings_t
|
||||
|
||||
|
||||
class cache_cli_t:
|
||||
@staticmethod
|
||||
def add_arguments(parser: argparse.ArgumentParser) -> None:
|
||||
default_dir = cache_settings_t.model_fields['dir'].default
|
||||
parser.add_argument(
|
||||
'--cache-dir',
|
||||
dest='cache_dir',
|
||||
default=None,
|
||||
help='directory for cached .db files and sqlite database (default: %s)' % default_dir,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def extract(namespace: argparse.Namespace) -> dict[str, Any]:
|
||||
kwargs: dict[str, Any] = {}
|
||||
if getattr(namespace, 'cache_dir', None) is not None:
|
||||
kwargs['dir'] = namespace.cache_dir
|
||||
return kwargs
|
||||
|
||||
@staticmethod
|
||||
def apply(kwargs: dict[str, Any]) -> cache_settings_t:
|
||||
if len(kwargs) > 0:
|
||||
return cache_settings_t.reset(**kwargs)
|
||||
return cache_settings_t.singleton()
|
||||
32
python/online/fxreader/pr34/commands_typed/archlinux/apps/cache/settings.py
vendored
Normal file
32
python/online/fxreader/pr34/commands_typed/archlinux/apps/cache/settings.py
vendored
Normal file
@ -0,0 +1,32 @@
|
||||
"""Cache settings singleton based on pydantic-settings.
|
||||
|
||||
Values can be set via environment variables (ARCHLINUX_CACHE_DIR, etc.)
|
||||
or by calling cache_settings_t.reset() with explicit kwargs.
|
||||
"""
|
||||
|
||||
import pathlib
|
||||
|
||||
import pydantic_settings
|
||||
|
||||
from typing import Any, ClassVar, Optional
|
||||
|
||||
|
||||
class cache_settings_t(pydantic_settings.BaseSettings):
|
||||
model_config = pydantic_settings.SettingsConfigDict(
|
||||
env_prefix='ARCHLINUX_CACHE_',
|
||||
)
|
||||
|
||||
dir: pathlib.Path = pathlib.Path.home() / '.cache' / 'online.fxreader.pr34.commands_typed.archlinux' / 'apps' / 'cache' / 'cache_dir'
|
||||
|
||||
_instance: ClassVar[Optional['cache_settings_t']] = None
|
||||
|
||||
@classmethod
|
||||
def singleton(cls) -> 'cache_settings_t':
|
||||
if cls._instance is None:
|
||||
cls._instance = cls()
|
||||
return cls._instance
|
||||
|
||||
@classmethod
|
||||
def reset(cls, **kwargs: Any) -> 'cache_settings_t':
|
||||
cls._instance = cls.model_validate(kwargs)
|
||||
return cls._instance
|
||||
@ -15,6 +15,7 @@ from typing import (
|
||||
)
|
||||
|
||||
from ..apps.cache.db import cache_db_t
|
||||
from ..apps.specs.utils import parse_reference
|
||||
from .archive_types import manager_t
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -44,12 +45,6 @@ def main(args: list[str]) -> int:
|
||||
'action',
|
||||
choices=[o.value for o in ArchiveAction],
|
||||
)
|
||||
archive_parser.add_argument(
|
||||
'--cache-dir',
|
||||
dest='cache_dir',
|
||||
required=True,
|
||||
help='directory for cached .db files and sqlite database',
|
||||
)
|
||||
archive_parser.add_argument(
|
||||
'--manager',
|
||||
default='pacman',
|
||||
@ -89,6 +84,12 @@ def main(args: list[str]) -> int:
|
||||
default=None,
|
||||
help='package names for show-versions, comma-separated',
|
||||
)
|
||||
archive_parser.add_argument(
|
||||
'--reference',
|
||||
default=None,
|
||||
help='path to compiled requirements file; '
|
||||
'sync will fetch archive dates for pinned versions not yet in cache',
|
||||
)
|
||||
|
||||
archive_options = archive_parser.parse_args(args)
|
||||
archive_options.action = ArchiveAction(archive_options.action)
|
||||
@ -100,7 +101,9 @@ def main(args: list[str]) -> int:
|
||||
if p.strip()
|
||||
]
|
||||
|
||||
cache_dir = pathlib.Path(archive_options.cache_dir)
|
||||
from ..apps.cache.settings import cache_settings_t
|
||||
|
||||
cache_dir = cache_settings_t.singleton().dir
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
db = cache_db_t(cache_dir / 'archlinux_cache.db')
|
||||
@ -139,7 +142,17 @@ def main(args: list[str]) -> int:
|
||||
elif archive_options.action is ArchiveAction.sync:
|
||||
mgr = _get_manager(archive_options.manager)
|
||||
|
||||
if archive_options.date is not None:
|
||||
if archive_options.reference is not None:
|
||||
ref_txt = pathlib.Path(archive_options.reference).read_text()
|
||||
ref_pinned = parse_reference(ref_txt)
|
||||
mgr.sync_reference(
|
||||
reference=ref_pinned,
|
||||
cache_dir=cache_dir,
|
||||
cache_db=db,
|
||||
repos=archive_options.repos,
|
||||
arch=archive_options.arch,
|
||||
)
|
||||
elif archive_options.date is not None:
|
||||
mgr.sync_date(
|
||||
date=archive_options.date,
|
||||
cache_dir=cache_dir,
|
||||
@ -158,7 +171,7 @@ def main(args: list[str]) -> int:
|
||||
step_days=archive_options.date_step,
|
||||
)
|
||||
else:
|
||||
logger.error('sync requires --date or --date-range')
|
||||
logger.error('sync requires --date, --date-range, or --reference')
|
||||
return 1
|
||||
else:
|
||||
raise NotImplementedError
|
||||
|
||||
@ -0,0 +1,94 @@
|
||||
import argparse
|
||||
import pathlib
|
||||
import unittest
|
||||
import unittest.mock
|
||||
|
||||
from ..apps.cache.settings import cache_settings_t
|
||||
from ..apps.cache.cli import cache_cli_t
|
||||
|
||||
|
||||
class TestCacheSettings(unittest.TestCase):
|
||||
def tearDown(self) -> None:
|
||||
cache_settings_t._instance = None
|
||||
|
||||
def test_singleton_returns_same_instance(self) -> None:
|
||||
a = cache_settings_t.singleton()
|
||||
b = cache_settings_t.singleton()
|
||||
self.assertIs(a, b)
|
||||
|
||||
def test_default_dir(self) -> None:
|
||||
s = cache_settings_t.singleton()
|
||||
expected = (
|
||||
pathlib.Path.home()
|
||||
/ '.cache'
|
||||
/ 'online.fxreader.pr34.commands_typed.archlinux'
|
||||
/ 'apps'
|
||||
/ 'cache'
|
||||
/ 'cache_dir'
|
||||
)
|
||||
self.assertEqual(s.dir, expected)
|
||||
|
||||
def test_reset_changes_dir(self) -> None:
|
||||
cache_settings_t.reset(dir='/tmp/test_cache')
|
||||
s = cache_settings_t.singleton()
|
||||
self.assertEqual(s.dir, pathlib.Path('/tmp/test_cache'))
|
||||
|
||||
def test_reset_returns_new_instance(self) -> None:
|
||||
a = cache_settings_t.singleton()
|
||||
b = cache_settings_t.reset(dir='/tmp/other')
|
||||
self.assertIsNot(a, b)
|
||||
|
||||
def test_env_override(self) -> None:
|
||||
with unittest.mock.patch.dict('os.environ', {'ARCHLINUX_CACHE_DIR': '/tmp/env_cache'}):
|
||||
s = cache_settings_t()
|
||||
self.assertEqual(s.dir, pathlib.Path('/tmp/env_cache'))
|
||||
|
||||
|
||||
class TestCacheCli(unittest.TestCase):
|
||||
def tearDown(self) -> None:
|
||||
cache_settings_t._instance = None
|
||||
|
||||
def test_add_arguments(self) -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
cache_cli_t.add_arguments(parser)
|
||||
ns = parser.parse_args(['--cache-dir', '/tmp/my_cache'])
|
||||
self.assertEqual(ns.cache_dir, '/tmp/my_cache')
|
||||
|
||||
def test_add_arguments_default_is_none(self) -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
cache_cli_t.add_arguments(parser)
|
||||
ns = parser.parse_args([])
|
||||
self.assertIsNone(ns.cache_dir)
|
||||
|
||||
def test_extract(self) -> None:
|
||||
ns = argparse.Namespace(cache_dir='/tmp/extracted')
|
||||
kwargs = cache_cli_t.extract(ns)
|
||||
self.assertEqual(kwargs, {'dir': '/tmp/extracted'})
|
||||
|
||||
def test_extract_empty(self) -> None:
|
||||
ns = argparse.Namespace(cache_dir=None)
|
||||
kwargs = cache_cli_t.extract(ns)
|
||||
self.assertEqual(kwargs, {})
|
||||
|
||||
def test_apply_with_override(self) -> None:
|
||||
s = cache_cli_t.apply({'dir': '/tmp/applied'})
|
||||
self.assertEqual(s.dir, pathlib.Path('/tmp/applied'))
|
||||
self.assertIs(s, cache_settings_t.singleton())
|
||||
|
||||
def test_apply_empty_returns_default(self) -> None:
|
||||
s = cache_cli_t.apply({})
|
||||
self.assertEqual(
|
||||
s.dir,
|
||||
pathlib.Path.home()
|
||||
/ '.cache'
|
||||
/ 'online.fxreader.pr34.commands_typed.archlinux'
|
||||
/ 'apps'
|
||||
/ 'cache'
|
||||
/ 'cache_dir',
|
||||
)
|
||||
|
||||
def test_help_contains_default_path(self) -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
cache_cli_t.add_arguments(parser)
|
||||
help_text = parser.format_help()
|
||||
self.assertIn('apps/cache/cache_dir', help_text)
|
||||
Loading…
Reference in New Issue
Block a user