Merge pull request #3 from ffaraone/rich_click

Add rich click

pyproject.toml

@@ -39,6 +39,7 @@ rich = "^13.7.0"
 pyyaml = "^6.0.1"
 python-telegram-bot = "^20.8"
 humanize = "^4.9.0"
+rich-click = "^1.7.3"
 
 [tool.poetry.scripts]
 sud = "sud.cli:main"

sud/cli.py

@@ -1,55 +1,65 @@
 import logging
 
-import click
-from click import Abort, ClickException, Option, UsageError
 from rich.console import Console
 from rich.logging import RichHandler
 
-from sud import get_version
-from sud.config import Config
+from sud import click as click
+from sud import config, get_version
 from sud.updater import Updater
 
 console = Console()
 
 
-class MutuallyExclusiveOption(Option):
-    def __init__(self, *args, **kwargs):
-        self.mutually_exclusive = set(kwargs.pop("mutually_exclusive", []))
-        help = kwargs.get("help", "")
-        if self.mutually_exclusive:
-            ex_str = ", ".join(self.mutually_exclusive)
-            kwargs["help"] = help + (
-                " NOTE: This option is mutually exclusive "
-                f"with options: [{ex_str}]."
-            )
-        super(MutuallyExclusiveOption, self).__init__(*args, **kwargs)
-
-    def handle_parse_result(self, ctx, opts, args):
-        if self.mutually_exclusive.intersection(opts) and self.name in opts:
-            raise UsageError(
-                f"Illegal usage: `{self.name}` is mutually exclusive with "
-                f"options `{', '.join(self.mutually_exclusive)}`."
-            )
-
-
 @click.group()
 @click.option(
     "-c",
     "--config-file",
-    type=click.File("rb"),
+    type=click.Path(
+        dir_okay=False,
+        readable=True,
+        writable=True,
+    ),
     default="/etc/sud/sud-config.yml",
+    help="Load the configuration file from a specific location.",
 )
 @click.version_option(get_version())
-@click.pass_context
+@config.pass_config
 def cli(ctx, config_file):
     """
     SUD the Python Scaleway DNS Updater utility.
     """
-    ctx.obj = Config(config_file)
+
+
+@cli.command(load_config=False)
+# @click.option(
+#     "-H",
+#     "--hostname",
+#     required=True,
+#     prompt="[cyan]Hostname[/cyan]",
+# )
+# @click.option(
+#     "-s",
+#     "--api-secret",
+#     required=True,
+#     prompt="[cyan]Scaleway API secret[/cyan]",
+#     hide_input=True,
+#     confirmation_prompt=True,
+# )
+@click.option(
+    "-s",
+    "--frequency",
+    required=True,
+    prompt="Check frequency",
+    default=300,
+    show_default=True,
+)
+@config.pass_config
+def init(config, frequency):
+    pass
 
 
 @cli.command()
-@click.pass_obj
+@config.pass_config
 def run(config):
     logging.basicConfig(
         level=logging.INFO,
@@ -64,9 +74,9 @@ def run(config):
 def main():
     try:
         cli(standalone_mode=False)
-    except ClickException as e:
+    except click.ClickException as e:
         console.print(f"[bold red]Error:[/bold red] {str(e)}")
-    except Abort:
+    except click.Abort:
         pass
     except Exception:
         console.print_exception()

sud/click.py

@@ -0,0 +1,160 @@
+# flake8: noqa: F401
+from typing import Any, Sequence
+
+import rich_click
+from click.types import ParamType
+from rich_click import Context as Context
+from rich_click import Abort as Abort
+from rich_click import ClickException as ClickException
+from rich_click import Path as Path
+from rich_click import version_option as version_option
+
+from sud.prompt import confirm, prompt
+
+
+class SudOption(rich_click.Option):
+    def __init__(
+        self,
+        param_decls: Sequence[str] | None = None,
+        show_default: bool | str | None = None,
+        prompt: bool | str = False,
+        confirmation_prompt: bool | str = False,
+        prompt_required: bool = True,
+        hide_input: bool = False,
+        is_flag: bool | None = None,
+        flag_value: Any | None = None,
+        multiple: bool = False,
+        count: bool = False,
+        allow_from_autoenv: bool = True,
+        type: ParamType | Any | None = None,
+        help: str | None = None,
+        hidden: bool = False,
+        show_choices: bool = True,
+        show_envvar: bool = False,
+        **attrs: Any,
+    ) -> None:
+        self.default_is_missing = "default" not in attrs
+        super().__init__(
+            param_decls,
+            show_default,
+            prompt,
+            confirmation_prompt,
+            prompt_required,
+            hide_input,
+            is_flag,
+            flag_value,
+            multiple,
+            count,
+            allow_from_autoenv,
+            type,
+            help,
+            hidden,
+            show_choices,
+            show_envvar,
+            **attrs,
+        )
+
+    def prompt_for_value(self, ctx: Context) -> Any:
+        assert self.prompt is not None
+
+        # Calculate the default before prompting anything to be stable.
+        default = self.get_default(ctx)
+        # If this is a prompt for a flag we need to handle this
+        # differently.
+        if self.is_bool_flag:
+            return confirm(self.prompt, default)
+
+        return prompt(
+            self.prompt,
+            default=default,
+            type=self.type,
+            hide_input=self.hide_input,
+            show_choices=self.show_choices,
+            confirmation_prompt=self.confirmation_prompt,
+            value_proc=lambda x: self.process_value(ctx, x),
+            default_is_missing=self.default_is_missing,
+        )
+
+
+class MutuallyExclusiveOption(SudOption):
+    def __init__(self, *args, **kwargs):
+        self.mutually_exclusive = set(kwargs.pop("mutually_exclusive", []))
+        help = kwargs.get("help", "")
+        if self.mutually_exclusive:
+            ex_str = ", ".join(self.mutually_exclusive)
+            kwargs["help"] = help + (
+                " NOTE: This option is mutually exclusive "
+                f"with options: [{ex_str}]."
+            )
+        super(MutuallyExclusiveOption, self).__init__(*args, **kwargs)
+
+    def handle_parse_result(self, ctx, opts, args):
+        if self.mutually_exclusive.intersection(opts) and self.name in opts:
+            raise rich_click.UsageError(
+                f"Illegal usage: `{self.name}` is mutually exclusive with "
+                f"options `{', '.join(self.mutually_exclusive)}`."
+            )
+
+
+class SudCommand(rich_click.Command):
+    def __init__(self, *args, **kwargs):
+        self.load_config = kwargs.pop("load_config", True)
+        super().__init__(*args, **kwargs)
+
+    def invoke(self, ctx):
+        if self.load_config:
+            ctx.obj.load()
+        return super().invoke(ctx)
+
+
+class SudGroup(rich_click.Group):
+    def command(self, *args, **kwargs):
+        from rich_click.decorators import command
+
+        kwargs["cls"] = SudCommand
+
+        def decorator(f):
+            cmd = command(*args, **kwargs)(f)
+            self.add_command(cmd)
+            return cmd
+
+        return decorator
+
+    def group(self, *args, **kwargs):
+        from rich_click.decorators import group
+
+        kwargs["cls"] = SudGroup
+
+        def decorator(f):
+            cmd = group(*args, **kwargs)(f)
+            self.add_command(cmd)
+            return cmd
+
+        return decorator
+
+
+def group(name=None, **attrs):
+    attrs.setdefault("cls", SudGroup)
+    return rich_click.command(name, **attrs)
+
+
+def option(
+    *param_decls: str,
+    **attrs: Any,
+):
+    attrs.setdefault("cls", SudOption)
+    return rich_click.option(
+        *param_decls,
+        **attrs,
+    )
+
+
+def mutually_exclusive_option(
+    *param_decls: str,
+    **attrs: Any,
+):
+    attrs.setdefault("cls", MutuallyExclusiveOption)
+    return rich_click.option(
+        *param_decls,
+        **attrs,
+    )

sud/config.py

@@ -1,6 +1,8 @@
 from datetime import timedelta
+from functools import update_wrapper
 
 import yaml
+from click import pass_context
 
 from sud.exceptions import SudException
 
@@ -10,8 +12,8 @@ class Config:
     DEFAULT_FREQUENCY = 300
 
     def __init__(self, config_file):
-        self._config = self._load_config(config_file)
-        self._validate_config()
+        self._config_file = config_file
+        self._config = None
 
     @property
     def hostname(self) -> str:
@@ -33,13 +35,25 @@ def api_secret(self) -> str:
     def telegram(self) -> dict | None:
         return self._config.get("notifications", {}).get("telegram")
 
-    def _load_config(self, config_file):
+    def load(self):
         try:
-            return yaml.safe_load(config_file)
+            with open(self._config_file, "r") as f:
+                self._config = yaml.safe_load(f)
         except yaml.YAMLError as e:
             raise SudException(
                 f"Invalid SUD configuration file: {str(e)}",
             ) from e
 
-    def _validate_config(self):
+    def validate(self):
         pass
+
+
+def pass_config(f):
+    @pass_context
+    def new_func(ctx, *args, **kwargs):
+        obj = ctx.find_object(Config)
+        if not obj:
+            ctx.obj = obj = Config(ctx.params["config_file"])
+        return ctx.invoke(f, obj, *args, **kwargs)
+
+    return update_wrapper(new_func, f)

sud/exceptions.py

@@ -1,6 +1,5 @@
-class SudException(Exception):
-    def __init__(self, message):
-        self.message = message
+from rich_click import ClickException
 
-    def __str__(self) -> str:
-        return self.message
+
+class SudException(ClickException):
+    pass