Add SettingOption to bridge Click CLI options to Plain settings

CLI options using envvar= bypass Plain's settings system (no prefix
support, invisible to `plain settings list`). SettingOption reads from
Plain settings instead, making the setting the single source of truth
for defaults and env var resolution.

Author: davegaeddert

plain-jobs/plain/jobs/cli.py

@@ -7,7 +7,7 @@
 
 import click
 
-from plain.cli import register_cli
+from plain.cli import SettingOption, register_cli
 from plain.runtime import settings
 from plain.utils import timezone
 
@@ -37,30 +37,30 @@ def cli() -> None:
 @click.option(
     "--max-processes",
     "max_processes",
-    default=None,
     type=int,
-    envvar="PLAIN_JOBS_WORKER_MAX_PROCESSES",
+    cls=SettingOption,
+    setting="JOBS_WORKER_MAX_PROCESSES",
 )
 @click.option(
     "--max-jobs-per-process",
     "max_jobs_per_process",
-    default=None,
     type=int,
-    envvar="PLAIN_JOBS_WORKER_MAX_JOBS_PER_PROCESS",
+    cls=SettingOption,
+    setting="JOBS_WORKER_MAX_JOBS_PER_PROCESS",
 )
 @click.option(
     "--max-pending-per-process",
     "max_pending_per_process",
-    default=10,
     type=int,
-    envvar="PLAIN_JOBS_WORKER_MAX_PENDING_PER_PROCESS",
+    cls=SettingOption,
+    setting="JOBS_WORKER_MAX_PENDING_PER_PROCESS",
 )
 @click.option(
     "--stats-every",
     "stats_every",
-    default=60,
     type=int,
-    envvar="PLAIN_JOBS_WORKER_STATS_EVERY",
+    cls=SettingOption,
+    setting="JOBS_WORKER_STATS_EVERY",
 )
 @click.option(
     "--reload",
@@ -78,64 +78,59 @@ def worker(
     """Run the job worker"""
     jobs_schedule = load_schedule(settings.JOBS_SCHEDULE)
 
-    if reload:
-        from plain.internal.reloader import Reloader
-
-        # Track whether we should continue restarting
-        should_restart = {"value": True}
-        current_worker = {"instance": None}
-
-        def file_changed(filename: str) -> None:
-            if current_worker["instance"]:
-                current_worker["instance"].shutdown()
-
-        def signal_shutdown(signalnum: int, _: Any) -> None:
-            should_restart["value"] = False
-            if current_worker["instance"]:
-                current_worker["instance"].shutdown()
-
-        # Allow the worker to be stopped gracefully on SIGTERM/SIGINT
-        signal.signal(signal.SIGTERM, signal_shutdown)
-        signal.signal(signal.SIGINT, signal_shutdown)
-
-        # Start file watcher once, outside the loop
-        reloader = Reloader(callback=file_changed, watch_html=False)
-        reloader.start()
-
-        while should_restart["value"]:
-            worker = Worker(
-                queues=list(queues),
-                jobs_schedule=jobs_schedule,
-                max_processes=max_processes,
-                max_jobs_per_process=max_jobs_per_process,
-                max_pending_per_process=max_pending_per_process,
-                stats_every=stats_every,
-            )
-            current_worker["instance"] = worker
-
-            # Start processing jobs (blocks until shutdown)
-            worker.run()
+    worker_kwargs = {
+        "queues": list(queues),
+        "jobs_schedule": jobs_schedule,
+        "max_processes": max_processes,
+        "max_jobs_per_process": max_jobs_per_process,
+        "max_pending_per_process": max_pending_per_process,
+        "stats_every": stats_every,
+    }
 
+    if reload:
+        _run_with_reload(worker_kwargs)
     else:
-        worker = Worker(
-            queues=list(queues),
-            jobs_schedule=jobs_schedule,
-            max_processes=max_processes,
-            max_jobs_per_process=max_jobs_per_process,
-            max_pending_per_process=max_pending_per_process,
-            stats_every=stats_every,
-        )
+        _run_once(worker_kwargs)
+
+
+def _run_with_reload(worker_kwargs: dict[str, Any]) -> None:
+    from plain.internal.reloader import Reloader
+
+    should_restart = {"value": True}
+    current_worker: dict[str, Worker | None] = {"instance": None}
+
+    def file_changed(filename: str) -> None:
+        if current_worker["instance"]:
+            current_worker["instance"].shutdown()
+
+    def signal_shutdown(signalnum: int, _: Any) -> None:
+        should_restart["value"] = False
+        if current_worker["instance"]:
+            current_worker["instance"].shutdown()
+
+    signal.signal(signal.SIGTERM, signal_shutdown)
+    signal.signal(signal.SIGINT, signal_shutdown)
+
+    reloader = Reloader(callback=file_changed, watch_html=False)
+    reloader.start()
+
+    while should_restart["value"]:
+        w = Worker(**worker_kwargs)
+        current_worker["instance"] = w
+        w.run()
+
+
+def _run_once(worker_kwargs: dict[str, Any]) -> None:
+    w = Worker(**worker_kwargs)
 
-        def _shutdown(signalnum: int, _: Any) -> None:
-            logger.info("Job worker shutdown signal received signalnum=%s", signalnum)
-            worker.shutdown()
+    def _shutdown(signalnum: int, _: Any) -> None:
+        logger.info("Job worker shutdown signal received signalnum=%s", signalnum)
+        w.shutdown()
 
-        # Allow the worker to be stopped gracefully on SIGTERM
-        signal.signal(signal.SIGTERM, _shutdown)
-        signal.signal(signal.SIGINT, _shutdown)
+    signal.signal(signal.SIGTERM, _shutdown)
+    signal.signal(signal.SIGINT, _shutdown)
 
-        # Start processing jobs
-        worker.run()
+    w.run()
 
 
 @cli.command()
@@ -195,8 +190,7 @@ def run(job_class_name: str) -> None:
 def list_jobs() -> None:
     """List all registered jobs"""
     for name, job_class in jobs_registry.jobs.items():
-        click.secho(f"{name}", bold=True, nl=False)
-        # Get description from class docstring
+        click.secho(name, bold=True, nl=False)
         description = job_class.__doc__.strip() if job_class.__doc__ else ""
         if description:
             click.secho(f": {description}", dim=True)

plain-jobs/plain/jobs/default_settings.py

@@ -4,3 +4,7 @@
     "plain.jobs.middleware.AppLoggerMiddleware",
 ]
 JOBS_SCHEDULE: list[tuple[str, str]] = []
+JOBS_WORKER_MAX_PROCESSES: int | None = None
+JOBS_WORKER_MAX_JOBS_PER_PROCESS: int | None = None
+JOBS_WORKER_MAX_PENDING_PER_PROCESS: int = 10
+JOBS_WORKER_STATS_EVERY: int = 60

plain/plain/cli/__init__.py

@@ -1,4 +1,5 @@
+from .options import SettingOption
 from .registry import register_cli
 from .runtime import common_command
 
-__all__ = ["register_cli", "common_command"]
+__all__ = ["SettingOption", "register_cli", "common_command"]

plain/plain/cli/options.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any
+
+import click
+
+
+class SettingOption(click.Option):
+    """A Click option that reads its value from Plain settings instead of os.environ.
+
+    Usage:
+        @click.option("--max-processes", cls=SettingOption, setting="JOBS_WORKER_MAX_PROCESSES")
+
+    CLI args always take priority. The setting's value (including package
+    defaults) is the single source of truth — no need for Click `default=`.
+    """
+
+    def __init__(self, *args: Any, setting: str | None = None, **kwargs: Any) -> None:
+        if setting and kwargs.get("envvar"):
+            raise ValueError(
+                "Cannot use both 'setting' and 'envvar' on the same option. "
+                "Use 'setting' to read from Plain settings."
+            )
+        if setting and "default" in kwargs:
+            raise ValueError(
+                "Cannot use both 'setting' and 'default' on the same option. "
+                "The setting's default value is the single source of truth."
+            )
+        self.setting_name = setting
+        if setting:
+            kwargs.setdefault("show_default", True)
+        super().__init__(*args, **kwargs)
+
+    def get_default(self, ctx: click.Context, call: bool = True) -> Any:
+        if self.setting_name:
+            try:
+                from plain.runtime import settings
+
+                settings._setup()
+                defn = settings._settings.get(self.setting_name)
+                if defn is not None:
+                    return defn.value
+            except Exception:
+                if ctx.resilient_parsing:
+                    return None
+                raise
+        return super().get_default(ctx, call=call)
+
+    def get_help_record(self, ctx: click.Context) -> tuple[str, str] | None:
+        result = super().get_help_record(ctx)
+        if result and self.setting_name:
+            name, help_text = result
+            setting_str = f"setting: {self.setting_name}"
+            if help_text and help_text.rstrip().endswith("]"):
+                bracket_start = help_text.rindex("[")
+                inside = help_text[bracket_start + 1 : -1]
+                help_text = f"{help_text[:bracket_start]}[{setting_str}; {inside}]"
+            elif help_text:
+                help_text = f"{help_text}  [{setting_str}]"
+            else:
+                help_text = f"[{setting_str}]"
+            return name, help_text
+        return result

plain/tests/app/test/default_settings.py

@@ -2,3 +2,4 @@
 EXPLICIT_SETTING: str = "unchanged explicit"
 ENV_SETTING: int = 0
 ENV_OVERRIDDEN_SETTING: str = "unchanged env overridden"
+NULLABLE_SETTING: int | None = None

plain/tests/test_setting_option.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import click
+import pytest
+from click.testing import CliRunner
+
+from plain.cli.options import SettingOption
+
+
+@click.command()
+@click.option(
+    "--value",
+    type=int,
+    cls=SettingOption,
+    setting="ENV_SETTING",
+)
+def sample_cmd(value):
+    click.echo(f"value={value}")
+
+
+def test_cli_arg_overrides_setting():
+    runner = CliRunner()
+    # ENV_SETTING is set to 1 via conftest.py env var
+    result = runner.invoke(sample_cmd, ["--value", "42"])
+    assert result.exit_code == 0
+    assert "value=42" in result.output
+
+
+def test_setting_from_env_var():
+    runner = CliRunner()
+    # ENV_SETTING = 1 (set via PLAIN_ENV_SETTING in conftest.py)
+    result = runner.invoke(sample_cmd, [])
+    assert result.exit_code == 0
+    assert "value=1" in result.output
+
+
+def test_none_when_setting_not_found():
+    @click.command()
+    @click.option(
+        "--value",
+        type=int,
+        cls=SettingOption,
+        setting="NONEXISTENT_SETTING_XYZ",
+    )
+    def cmd(value):
+        click.echo(f"value={value}")
+
+    runner = CliRunner()
+    result = runner.invoke(cmd, [])
+    assert result.exit_code == 0
+    assert "value=None" in result.output
+
+
+def test_setting_from_explicit_value():
+    @click.command()
+    @click.option(
+        "--value",
+        type=str,
+        cls=SettingOption,
+        setting="EXPLICIT_SETTING",
+    )
+    def cmd(value):
+        click.echo(f"value={value}")
+
+    runner = CliRunner()
+    result = runner.invoke(cmd, [])
+    assert result.exit_code == 0
+    assert "value=explicitly changed" in result.output
+
+
+def test_raises_if_both_setting_and_envvar():
+    with pytest.raises(ValueError, match="Cannot use both"):
+
+        @click.command()
+        @click.option(
+            "--value",
+            type=int,
+            cls=SettingOption,
+            setting="SOME_SETTING",
+            envvar="SOME_ENV_VAR",
+        )
+        def cmd(value):
+            pass
+
+
+def test_raises_if_both_setting_and_default():
+    with pytest.raises(ValueError, match="Cannot use both"):
+
+        @click.command()
+        @click.option(
+            "--value",
+            type=int,
+            cls=SettingOption,
+            setting="SOME_SETTING",
+            default=42,
+        )
+        def cmd(value):
+            pass
+
+
+def test_none_valued_setting():
+    """A setting that exists but has None as its value passes through as None."""
+
+    @click.command()
+    @click.option(
+        "--value",
+        type=int,
+        cls=SettingOption,
+        setting="NULLABLE_SETTING",
+    )
+    def cmd(value):
+        click.echo(f"value={value}")
+
+    runner = CliRunner()
+    result = runner.invoke(cmd, [])
+    assert result.exit_code == 0
+    assert "value=None" in result.output
+
+
+def test_package_default_is_used():
+    """Package defaults flow through SettingOption as the single source of truth."""
+
+    @click.command()
+    @click.option(
+        "--value",
+        type=str,
+        cls=SettingOption,
+        setting="DEFAULT_SETTING",
+    )
+    def cmd(value):
+        click.echo(f"value={value}")
+
+    runner = CliRunner()
+    result = runner.invoke(cmd, [])
+    assert result.exit_code == 0
+    assert "value=unchanged default" in result.output