From e62943b5a46d9fac76d35a27e6524704fcf8e58f Mon Sep 17 00:00:00 2001 From: Dzmitry Pryskoka Date: Sun, 21 Jan 2024 19:02:01 +0300 Subject: [PATCH] Refactor package CLI --- src/kataloger/__init__.py | 1 + src/kataloger/__main__.py | 152 +------------------- src/kataloger/cli/__init__.py | 0 src/kataloger/cli/cli.py | 38 +++++ src/kataloger/cli/configuration_provider.py | 107 ++++++++++++++ src/kataloger/cli/update_print_helper.py | 29 ++++ 6 files changed, 181 insertions(+), 146 deletions(-) create mode 100644 src/kataloger/cli/__init__.py create mode 100644 src/kataloger/cli/cli.py create mode 100644 src/kataloger/cli/configuration_provider.py create mode 100644 src/kataloger/cli/update_print_helper.py diff --git a/src/kataloger/__init__.py b/src/kataloger/__init__.py index e69de29..504740c 100644 --- a/src/kataloger/__init__.py +++ b/src/kataloger/__init__.py @@ -0,0 +1 @@ +package_name = "kataloger" diff --git a/src/kataloger/__main__.py b/src/kataloger/__main__.py index 79cd3c0..39545cc 100644 --- a/src/kataloger/__main__.py +++ b/src/kataloger/__main__.py @@ -1,160 +1,20 @@ import sys -from argparse import ArgumentParser -from importlib.metadata import version -from importlib.resources import files, as_file -from pathlib import Path -from typing import Optional import asyncio -from kataloger.catalog_updater_builder import CatalogUpdaterBuilder -from kataloger.data.kataloger_configuration import KatalogerConfiguration -from kataloger.execptions.kataloger_configuration_exception import KatalogerConfigurationException +from kataloger.cli import cli from kataloger.execptions.kataloger_exception import KatalogerException -from kataloger.update_resolver.universal.universal_update_resolver import UniversalUpdateResolver -from kataloger.update_resolver.universal.universal_version_factory import UniversalVersionFactory - - -def parse_arguments() -> KatalogerConfiguration: - parser = ArgumentParser( - prog="kataloger", - description="A Python command-line utility to discover updates for your gradle version catalogs.", - allow_abbrev=False, - epilog="Visit project repository to get more information." - ) - parser.add_argument( - "-p", "--path", - action="append", - dest="paths", - help="Path(s) to gradle version catalog. If catalog path not provided script looking for version catalogs in " - "current directory.", - ) - parser.add_argument( - "-rp", "--repositories-path", - type=str, - dest="repositories_path", - metavar="path", - help="Path to .toml file with repositories info. If repositories path not provided script looking for " - "default.repositories.toml file in current directory.", - ) - parser.add_argument( - "-v", "--verbose", - action="store_true", - dest="verbose", - help="Enables detailed output.", - ) - parser.add_argument( - "-u", "--suggest-unstable", - action="store_true", - dest="suggest_unstable_updates", - help="Allow %(prog)s suggest update from stable version to unstable.", - ) - parser.add_argument( - "-f", "--fail-on-updates", - action="store_true", - dest="fail_on_updates", - help="Exit with non-zero code when at least one update found.", - ) - parser.add_argument( - "--version", - action="version", - version=f"%(prog)s {get_kataloger_version()}", - ) - arguments = parser.parse_args() - - return KatalogerConfiguration( - catalogs=get_catalog_paths(arguments.paths), - repositories_path=get_repositories_path(arguments.repositories_path), - verbose=arguments.verbose, - suggest_unstable_updates=arguments.suggest_unstable_updates, - fail_on_updates=arguments.fail_on_updates, - ) - - -def get_kataloger_version() -> str: - if __name__ == '__main__': - return "indev" - else: - return version(__package__ or __name__) - - -def get_catalog_paths(path_strings: list[str]) -> list[Path]: - if path_strings: - return list(map(lambda path: str_to_path(path), path_strings)) - - # If catalog path not provided try to find catalogs in cwd. - catalog_files = Path.cwd().glob("*.versions.toml") - catalog_paths = list(filter(lambda path: path.exists() and path.is_file(), catalog_files)) - if not catalog_paths: - raise KatalogerConfigurationException("Gradle version catalog not found in current directory. Please specify " - "path to catalog via parameter or run tool from directory with catalog.") - - return catalog_paths - - -def get_repositories_path(path_string: Optional[str]) -> Path: - if path_string: - return str_to_path(path_string) - - repositories_candidate = Path.cwd() / "default.repositories.toml" - if repositories_candidate.exists() and repositories_candidate.is_file(): - return repositories_candidate - - with as_file(files("kataloger").joinpath("default.repositories.toml")) as path: - bundled_repositories_path = path - return bundled_repositories_path - - -def str_to_path(path_string: str) -> Path: - path = Path(path_string) - if not (path.exists() or path.is_file()): - raise KatalogerConfigurationException(message=f"Incorrect path: {path_string}.") - - return path def main() -> int: - async def async_main(): - configuration = parse_arguments() - - update_resolver = UniversalUpdateResolver( - version_factories=[UniversalVersionFactory()], - suggest_unstable_updates=configuration.suggest_unstable_updates, - ) - - catalog_updater = (CatalogUpdaterBuilder() - .add_resolver(update_resolver) - .set_repositories_path(configuration.repositories_path) - .set_verbose(verbose=configuration.verbose) - .build()) - - has_updates = False - for catalog_path in configuration.catalogs: - updates = await catalog_updater.get_catalog_updates(catalog_path) - if not has_updates and updates: - has_updates = True - - if len(configuration.catalogs) > 1 and updates: - if catalog_path.name.endswith(".versions.toml"): - catalog_name = catalog_path.name.removesuffix(".versions.toml") - else: - catalog_name = catalog_path.name - print(f"Updates for \"{catalog_name}\" catalog.") - for update in updates: - print(update) - if len(configuration.catalogs) > 1 and updates: - print("") - - if configuration.fail_on_updates and has_updates: - return 1 - else: - return 0 - try: - return asyncio.run(async_main()) + return asyncio.run(cli.run()) except KatalogerException as error: - print(f"Kataloger: {error.message}", file=sys.stderr) + print(error.message, file=sys.stderr) return 1 + except KeyboardInterrupt: + print("Update search terminated.") + return 128 + 2 # http://www.tldp.org/LDP/abs/html/exitcodes.html if __name__ == "__main__": diff --git a/src/kataloger/cli/__init__.py b/src/kataloger/cli/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/kataloger/cli/cli.py b/src/kataloger/cli/cli.py new file mode 100644 index 0000000..ff35800 --- /dev/null +++ b/src/kataloger/cli/cli.py @@ -0,0 +1,38 @@ +from kataloger.catalog_updater_builder import CatalogUpdaterBuilder +from kataloger.cli.configuration_provider import parse_configuration +from kataloger.cli.update_print_helper import print_catalog_updates +from kataloger.update_resolver.universal.universal_update_resolver import UniversalUpdateResolver +from kataloger.update_resolver.universal.universal_version_factory import UniversalVersionFactory + + +async def run() -> int: + configuration = parse_configuration() + + update_resolver = UniversalUpdateResolver( + version_factories=[UniversalVersionFactory()], + suggest_unstable_updates=configuration.suggest_unstable_updates, + ) + + catalog_updater = (CatalogUpdaterBuilder() + .add_resolver(update_resolver) + .set_repositories_path(configuration.repositories_path) + .set_verbose(verbose=configuration.verbose) + .build()) + + has_updates = False + for catalog_path in configuration.catalogs: + updates = await catalog_updater.get_catalog_updates(catalog_path) + if not has_updates and updates: + has_updates = True + + print_catalog_updates( + updates=updates, + catalog_name=catalog_path.name, + catalog_count=len(configuration.catalogs), + verbose=configuration.verbose, + ) + + if configuration.fail_on_updates and has_updates: + return 1 + else: + return 0 diff --git a/src/kataloger/cli/configuration_provider.py b/src/kataloger/cli/configuration_provider.py new file mode 100644 index 0000000..99b2aba --- /dev/null +++ b/src/kataloger/cli/configuration_provider.py @@ -0,0 +1,107 @@ +from argparse import ArgumentParser +from importlib.metadata import version +from importlib.resources import files, as_file +from pathlib import Path +from typing import Optional + +from kataloger import package_name +from kataloger.data.kataloger_configuration import KatalogerConfiguration +from kataloger.execptions.kataloger_configuration_exception import KatalogerConfigurationException + + +def parse_configuration() -> KatalogerConfiguration: + parser = ArgumentParser( + prog="kataloger", + description="A Python command-line utility to discover updates for your gradle version catalogs.", + allow_abbrev=False, + epilog="Visit project repository to get more information." + ) + parser.add_argument( + "-p", "--path", + action="append", + dest="paths", + help="Path(s) to gradle version catalog. If catalog path not provided script looking for version catalogs in " + "current directory.", + ) + parser.add_argument( + "-rp", "--repositories-path", + type=str, + dest="repositories_path", + metavar="path", + help="Path to .toml file with repositories info. If repositories path not provided script looking for " + "default.repositories.toml file in current directory.", + ) + parser.add_argument( + "-v", "--verbose", + action="store_true", + dest="verbose", + help="Enables detailed output.", + ) + parser.add_argument( + "-u", "--suggest-unstable", + action="store_true", + dest="suggest_unstable_updates", + help="Allow %(prog)s suggest update from stable version to unstable.", + ) + parser.add_argument( + "-f", "--fail-on-updates", + action="store_true", + dest="fail_on_updates", + help="Exit with non-zero code when at least one update found.", + ) + parser.add_argument( + "--version", + action="version", + version=f"%(prog)s {_get_kataloger_version()}", + ) + arguments = parser.parse_args() + + return KatalogerConfiguration( + catalogs=_get_catalog_paths(arguments.paths), + repositories_path=_get_repositories_path(arguments.repositories_path), + verbose=arguments.verbose, + suggest_unstable_updates=arguments.suggest_unstable_updates, + fail_on_updates=arguments.fail_on_updates, + ) + + +def _get_kataloger_version() -> str: + if __name__ == '__main__': + return "indev" + else: + return version(package_name) + + +def _get_catalog_paths(path_strings: list[str]) -> list[Path]: + if path_strings: + return list(map(lambda path: _str_to_path(path), path_strings)) + + # If catalog path not provided try to find catalogs in cwd. + catalog_files = Path.cwd().glob("*.versions.toml") + catalog_paths = list(filter(lambda path: path.exists() and path.is_file(), catalog_files)) + if not catalog_paths: + raise KatalogerConfigurationException("Gradle version catalog not found in current directory. Please specify " + "path to catalog via parameter or run tool from directory with catalog.") + + return catalog_paths + + +def _get_repositories_path(path_string: Optional[str]) -> Path: + if path_string: + return _str_to_path(path_string) + + repositories_candidate = Path.cwd() / "default.repositories.toml" + if repositories_candidate.exists() and repositories_candidate.is_file(): + return repositories_candidate + + with as_file(files(package_name).joinpath("default.repositories.toml")) as path: + bundled_repositories_path = path + return bundled_repositories_path + + +def _str_to_path(path_string: str) -> Path: + path = Path(path_string) + if not (path.exists() or path.is_file()): + raise KatalogerConfigurationException(message=f"Incorrect path: {path_string}.") + + return path diff --git a/src/kataloger/cli/update_print_helper.py b/src/kataloger/cli/update_print_helper.py new file mode 100644 index 0000000..24975a9 --- /dev/null +++ b/src/kataloger/cli/update_print_helper.py @@ -0,0 +1,29 @@ +from kataloger.data.artifact_update import ArtifactUpdate + + +def print_catalog_updates( + updates: list[ArtifactUpdate], + catalog_name: str, + catalog_count: int, + verbose: bool, +): + if catalog_name.endswith(".versions.toml"): + catalog_name = catalog_name.removesuffix(".versions.toml") + else: + catalog_name = catalog_name + + if catalog_count > 1: + if updates: + print(f"Updates for \"{catalog_name}\" catalog:") + else: + print(f"Catalog \"{catalog_name}\" is up to date!") + + for update in updates: + version_part = f"{update.current_version} -> {update.available_version}" + if verbose: + print(f"[{update.update_repository_name}] {update.name} {version_part}") + else: + print(f"{update.name} {version_part}") + + if catalog_count > 1: + print("")