rpc-server: use

invenio_cli/commands/assets.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2020 CERN.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Cli is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.

invenio_cli/commands/install.py

@@ -39,12 +39,12 @@ def install_py_dependencies(self, pre, dev=False):
     def update_instance_path(self):
         """Update path to instance in config."""
         result = run_cmd(
-            self.cli_config.python_package_manager.run_command(
+            self.cli_config.python_package_manager.send_command(
                 "invenio",
                 "shell",
                 "--no-term-title",
                 "-c",
-                "\"print(app.instance_path, end='')\"",
+                "print(app.instance_path, end='')",
             )
         )
         if result.status_code == 0:

invenio_cli/commands/local.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2020 CERN.
-# Copyright (C) 2022 Graz University of Technology.
+# Copyright (C) 2022-2025 Graz University of Technology.
 #
 # Invenio-Cli is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -84,16 +84,18 @@ def update_statics_and_assets(self, force, debug=False, log_file=None):
         Needed here (parent) because is used by Assets and Install commands.
         """
         # Commands
-        pkg_man = self.cli_config.python_package_manager
-        ops = [pkg_man.run_command("invenio", "collect", "--verbose")]
+        py_pkg_man = self.cli_config.python_package_manager
+
+        ops = [py_pkg_man.send_command("invenio", "collect", "--verbose")]
 
         if force:
-            ops.append(pkg_man.run_command("invenio", "webpack", "clean", "create"))
-            ops.append(pkg_man.run_command("invenio", "webpack", "install"))
+            ops.append(py_pkg_man.send_command("invenio", "webpack", "clean", "create"))
+            ops.append(py_pkg_man.send_command("invenio", "webpack", "install"))
         else:
-            ops.append(pkg_man.run_command("invenio", "webpack", "create"))
+            ops.append(py_pkg_man.send_command("invenio", "webpack", "create"))
         ops.append(self._statics)
-        ops.append(pkg_man.run_command("invenio", "webpack", "build"))
+        ops.append(py_pkg_man.send_command("invenio", "webpack", "build"))
+
         # Keep the same messages for some of the operations for backward compatibility
         messages = {
             "build": "Building assets...",

invenio_cli/helpers/cli_config.py

@@ -3,14 +3,15 @@
 # Copyright (C) 2019-2024 CERN.
 # Copyright (C) 2019-2020 Northwestern University.
 # Copyright (C) 2021 Esteban J. G. Gabancho.
-# Copyright (C) 2024 Graz University of Technology.
+# Copyright (C) 2024-2025 Graz University of Technology.
 #
 # Invenio-Cli is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Invenio-cli configuration file."""
 
 from configparser import ConfigParser
+from functools import cached_property
 from pathlib import Path
 
 from ..errors import InvenioCLIConfigError
@@ -62,7 +63,7 @@ def __init__(self, project_dir="./"):
             with open(self.private_config_path) as cfg_file:
                 self.private_config.read_file(cfg_file)
 
-    @property
+    @cached_property
     def python_package_manager(self) -> PythonPackageManager:
         """Get python packages manager."""
         manager_name = self.config[CLIConfig.CLI_SECTION].get(

invenio_cli/helpers/package_managers.py

@@ -1,22 +1,49 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2025 TU Wien.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Cli is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Wrappers around various package managers to be used under the hood."""
 
+import atexit
 import os
 from abc import ABC
+from subprocess import Popen
 from typing import List
 
+from .process import run_cmd
+
 
 class PythonPackageManager(ABC):
     """Interface for creating tool-specific Python package management commands."""
 
     name: str = None
     lock_file_name: str = None
+    rpc_server_is_running: bool = False
+    rpc_server: Popen = None
+    run_prefix: List = []
+
+    def __init__(self):
+        """Construct."""
+        self.rpc_server = Popen(
+            self.run_prefix + ["invenio", "rpc-server", "start", "--port", "5001"]
+        )
+        atexit.register(self.cleanup)
+        while True:
+            response = run_cmd(
+                self.run_prefix + ["rpc-server", "ping", "--port", "5001"]
+            )
+            if "pong" in response.output:
+                self.rpc_server_is_running = True
+                break
+
+    def cleanup(self):
+        """Cleanup."""
+        if self.rpc_server:
+            self.rpc_server.terminate()
 
     def run_command(self, *command: str) -> List[str]:
         """Generate command to run the given command in the managed environment."""
@@ -60,6 +87,24 @@ class Pipenv(PythonPackageManager):
 
     name = "pipenv"
     lock_file_name = "Pipfile.lock"
+    run_prefix = ["pipenv", "run"]
+
+    def send_command(self, *command):
+        """Send command to rpc server, default to run_command."""
+        if self.rpc_server_is_running:
+            # [1:] remove "invenio" from commands
+            return [
+                self.name,
+                "run",
+                "rpc-server",
+                "send",
+                "--port",
+                "5001",
+                "--plain",
+                *command[1:],
+            ]
+        else:
+            self.run_command(*command)
 
     def run_command(self, *command):
         """Generate command to run the given command in the managed environment."""
@@ -117,6 +162,25 @@ class UV(PythonPackageManager):
 
     name = "uv"
     lock_file_name = "uv.lock"
+    run_prefix = ["uv", "run", "--no-sync"]
+
+    def send_command(self, *command):
+        """Send command to rpc server, default to run_command."""
+        if self.rpc_server_is_running:
+            # [1:] remove "invenio" from commands
+            return [
+                self.name,
+                "run",
+                "--no-sync",
+                "rpc-server",
+                "send",
+                "--port",
+                "5001",
+                "--plain",
+                *command[1:],
+            ]
+        else:
+            return self.run_command(*command)
 
     def run_command(self, *command):
         """Generate command to run the given command in the managed environment."""