feat: Plugins are now auto-installed when commands require them

src/meltano/cli/add.py

@@ -9,19 +9,19 @@
 import click
 import requests
 
-from meltano.cli.params import pass_project
+from meltano.cli.params import InstallPlugins, install_option, pass_project
 from meltano.cli.utils import (
     CliError,
     PartialInstrumentedCmd,
     add_plugin,
     add_required_plugins,
     check_dependencies_met,
-    install_plugins,
 )
 from meltano.core.plugin import PluginRef, PluginType
 from meltano.core.plugin_install_service import PluginInstallReason
 from meltano.core.project_add_service import ProjectAddService
 from meltano.core.tracking.contexts import CliEvent, PluginsTrackingContext
+from meltano.core.utils import run_async
 from meltano.core.yaml import yaml
 
 if t.TYPE_CHECKING:
@@ -102,23 +102,21 @@ def _load_yaml_from_ref(_ctx, _param, value: str | None) -> dict | None:
     is_flag=True,
     help="Update an existing plugin.",
 )
-@click.option(
-    "--no-install",
-    is_flag=True,
-    help="Do not install the plugin after adding it to the project.",
-)
+@install_option
 @click.option(
     "--force-install",
     is_flag=True,
     help="Ignore the required Python version declared by the plugins.",
 )
 @pass_project()
 @click.pass_context
-def add(  # noqa: WPS238
+@run_async
+async def add(  # noqa: C901 WPS238
     ctx,
     project: Project,
     plugin_type: str,
     plugin_name: str,
+    install_plugins: InstallPlugins,
     inherit_from: str | None = None,
     variant: str | None = None,
     as_name: str | None = None,
@@ -197,17 +195,16 @@ def add(  # noqa: WPS238
     )
     tracker.track_command_event(CliEvent.inflight)
 
-    if not flags.get("no_install"):
-        success = install_plugins(
-            project,
-            plugins,
-            reason=PluginInstallReason.ADD,
-            force=flags.get("force_install", False),
-        )
+    success = await install_plugins(
+        project,
+        plugins,
+        reason=PluginInstallReason.ADD,
+        force=flags.get("force_install", False),
+    )
 
-        if not success:
-            tracker.track_command_event(CliEvent.failed)
-            raise CliError("Failed to install plugin(s)")  # noqa: EM101
+    if success is False:
+        tracker.track_command_event(CliEvent.failed)
+        raise CliError("Failed to install plugin(s)")  # noqa: EM101
 
     _print_plugins(plugins)
     tracker.track_command_event(CliEvent.completed)

src/meltano/cli/config.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import json
 import tempfile
 import typing as t
@@ -14,7 +13,7 @@
 import structlog
 
 from meltano.cli.interactive import InteractiveConfig
-from meltano.cli.params import pass_project
+from meltano.cli.params import InstallPlugins, install_option, pass_project
 from meltano.cli.utils import (
     CliEnvironmentBehavior,
     CliError,
@@ -25,11 +24,13 @@
 from meltano.core.plugin import PluginType
 from meltano.core.plugin.error import PluginNotFoundError
 from meltano.core.plugin.settings_service import PluginSettingsService
+from meltano.core.plugin_install_service import PluginInstallReason
 from meltano.core.plugin_invoker import PluginInvoker
 from meltano.core.plugin_test_service import PluginTestServiceFactory
 from meltano.core.settings_service import SettingValueStore
 from meltano.core.settings_store import StoreNotSupportedError
 from meltano.core.tracking.contexts import CliEvent, PluginsTrackingContext
+from meltano.core.utils import run_async
 
 if t.TYPE_CHECKING:
     from meltano.core.project import Project
@@ -393,8 +394,15 @@ def set_(
 
 
 @config.command(cls=PartialInstrumentedCmd, name="test")
+@pass_project(migrate=True)
 @click.pass_context
-def test(ctx):
+@install_option
+@run_async
+async def test(
+    ctx,
+    project: Project,
+    install_plugins: InstallPlugins,
+):
     """Test the configuration of a plugin."""
     invoker = ctx.obj["invoker"]
     tracker = ctx.obj["tracker"]
@@ -403,14 +411,18 @@ def test(ctx):
         raise CliError("Testing of the Meltano project configuration is not supported")  # noqa: EM101
 
     session = ctx.obj["session"]
+    plugin_test_service = PluginTestServiceFactory(invoker).get_test_service()
 
-    async def _validate():  # noqa: WPS430
-        plugin_test_service = PluginTestServiceFactory(invoker).get_test_service()
-        async with plugin_test_service.plugin_invoker.prepared(session):
-            return await plugin_test_service.validate()
+    await install_plugins(
+        project,
+        [plugin_test_service.plugin_invoker.plugin],
+        reason=PluginInstallReason.JIT,
+        skip_installed=True,
+    )
 
     try:
-        is_valid, detail = asyncio.run(_validate())
+        async with plugin_test_service.plugin_invoker.prepared(session):
+            is_valid, detail = await plugin_test_service.validate()
     except Exception:
         tracker.track_command_event(CliEvent.failed)
         raise

src/meltano/cli/elt.py

@@ -11,7 +11,7 @@
 import click
 from structlog import stdlib as structlog_stdlib
 
-from meltano.cli.params import pass_project
+from meltano.cli.params import InstallPlugins, install_option, pass_project
 from meltano.cli.utils import CliEnvironmentBehavior, CliError, PartialInstrumentedCmd
 from meltano.core.db import project_engine
 from meltano.core.elt_context import ELTContextBuilder
@@ -20,6 +20,7 @@
 from meltano.core.logging import JobLoggingService, OutputLogger
 from meltano.core.plugin import PluginType
 from meltano.core.plugin.error import PluginNotFoundError
+from meltano.core.plugin_install_service import PluginInstallReason
 from meltano.core.runner import RunnerError
 from meltano.core.runner.dbt import DbtRunner
 from meltano.core.runner.singer import SingerRunner
@@ -116,6 +117,7 @@ class ELOptions:
 @ELOptions.state_id
 @ELOptions.force
 @ELOptions.merge_state
+@install_option
 @click.pass_context
 @pass_project(migrate=True)
 @run_async
@@ -134,6 +136,7 @@ async def el(  # WPS408
     state_id: str,
     force: bool,
     merge_state: bool,
+    install_plugins: InstallPlugins,
 ):
     """
     Run an EL pipeline to Extract and Load data.
@@ -161,6 +164,7 @@ async def el(  # WPS408
         state_id,
         force,
         merge_state,
+        install_plugins,
     )
 
 
@@ -182,6 +186,7 @@ async def el(  # WPS408
 @ELOptions.state_id
 @ELOptions.force
 @ELOptions.merge_state
+@install_option
 @click.pass_context
 @pass_project(migrate=True)
 @run_async
@@ -201,6 +206,7 @@ async def elt(  # WPS408
     state_id: str,
     force: bool,
     merge_state: bool,
+    install_plugins: InstallPlugins,
 ):
     """
     Run an ELT pipeline to Extract, Load, and Transform data.
@@ -229,6 +235,7 @@ async def elt(  # WPS408
         state_id,
         force,
         merge_state,
+        install_plugins,
     )
 
 
@@ -248,6 +255,7 @@ async def _run_el_command(
     state_id: str,
     force: bool,
     merge_state: bool,
+    install_plugins: InstallPlugins,
 ):
     if platform.system() == "Windows":
         raise CliError(
@@ -294,7 +302,15 @@ async def _run_el_command(
         if dump:
             await dump_file(context_builder, dump)
         else:
-            await _run_job(tracker, project, job, session, context_builder, force=force)
+            await _run_job(
+                tracker,
+                project,
+                job,
+                session,
+                context_builder,
+                install_plugins,
+                force=force,
+            )
     except Exception as err:
         tracker.track_command_event(CliEvent.failed)
         raise err
@@ -358,7 +374,15 @@ async def dump_file(context_builder, dumpable):
         raise CliError(f"Could not dump {dumpable}: {err}") from err  # noqa: EM102
 
 
-async def _run_job(tracker, project, job, session, context_builder, force=False):
+async def _run_job(
+    tracker,
+    project,
+    job,
+    session,
+    context_builder,
+    install_plugins: InstallPlugins,
+    force=False,
+):
     fail_stale_jobs(session, job.job_name)
 
     if not force and (existing := JobFinder(job.job_name).latest_running(session)):
@@ -377,7 +401,7 @@ async def _run_job(tracker, project, job, session, context_builder, force=False)
 
         log = logger.bind(name="meltano", run_id=str(job.run_id), state_id=job.job_name)
 
-        await _run_elt(tracker, log, context_builder, output_logger)
+        await _run_elt(tracker, log, context_builder, output_logger, install_plugins)
 
 
 @asynccontextmanager
@@ -405,10 +429,23 @@ async def _run_elt(
     log: structlog.BoundLogger,
     context_builder: ELTContextBuilder,
     output_logger: OutputLogger,
+    install_plugins: InstallPlugins,
 ):
+    elt_context = context_builder.context()
+    plugins = [elt_context.extractor, elt_context.loader]
+
+    if elt_context.only_transform:
+        plugins.append(elt_context.transformer)
+
+    await install_plugins(
+        elt_context.project,
+        plugins,
+        reason=PluginInstallReason.JIT,
+        skip_installed=True,
+    )
+
     async with _redirect_output(log, output_logger):
         try:
-            elt_context = context_builder.context()
             tracker.add_contexts(PluginsTrackingContext.from_elt_context(elt_context))
             tracker.track_command_event(CliEvent.inflight)
 

src/meltano/cli/install.py

@@ -13,6 +13,7 @@
 from meltano.core.plugin import PluginType
 from meltano.core.schedule_service import ScheduleService
 from meltano.core.tracking.contexts import CliEvent, PluginsTrackingContext
+from meltano.core.utils import run_async
 
 if t.TYPE_CHECKING:
     from meltano.core.project import Project
@@ -59,7 +60,8 @@
 )
 @click.pass_context
 @pass_project(migrate=True)
-def install(  # noqa: C901
+@run_async
+async def install(  # noqa: C901
     project: Project,
     ctx: click.Context,
     plugin_type: str,
@@ -101,7 +103,7 @@ def install(  # noqa: C901
     )
     tracker.track_command_event(CliEvent.inflight)
 
-    success = install_plugins(
+    success = await install_plugins(
         project,
         plugins,
         parallelism=parallelism,