Merge pull request #31 from miguelaeh/plugins

feat(plugins): Rewrite plugin system

README.md

@@ -219,21 +219,8 @@ We provide some modules containing a growing set of ready to use models for comm
 
 ## Plugins
 
-We also provide a [plugins package](plugins) that provides useful plugins for common tasks such as producing events to Kafka, skipping image processing when they are similar, etc.
-
-To install the plugins package run:
-
-```console
-pip install pipeless-ai-plugins
-```
-
-To import it on your Pipeless project use:
-
-```python
-from pipeless_ai_plugins.<plugin_name> import <plugin_class>
-```
-
-To get more information about a specific plugin, including how to use it please check the specific plugin README.
+The Pipeless plugin system allows you to add functionality to your application out-of-the-box.
+Find the whole documentation about the Pipeless plugin system [here](https://pipeless.ai/docs/v0/plugins).
 
 ### Available plugins
 

cli/pyproject.toml

@@ -15,6 +15,8 @@ python = "^3.10"
 typer = {version = "^0.9.0", extras = ["all"]}
 pyyaml = "^6.0.1"
 pipeless-ai = "^0.1.0"
+gitpython = "^3.1.35"
+packaging = "^23.1"
 
 [build-system]
 requires = ["poetry-core"]

cli/src/pipeless_ai_cli/commands/create.py

@@ -44,6 +44,9 @@ def create_project(name: str):
         'worker': {
             'n_workers': 1,
         },
+        'plugins': {
+            'order': ''
+        }
     }
 
     try:

cli/src/pipeless_ai_cli/commands/install.py

@@ -0,0 +1,45 @@
+import os
+import sys
+import typer
+from rich import print as rprint
+
+from pipeless_ai_cli.lib.pip import install_pip_packages
+from pipeless_ai_cli.lib.plugins import get_latest_plugin_version_number, get_plugins_registry, download_plugin_from_repo_tag
+
+app = typer.Typer()
+
+@app.command(name="plugin", help="Install a plugin")
+def install_plugin(id: str, version: str = None, plugins_root: str = 'plugins'):
+    if not os.path.exists('app.py'):
+        rprint("[red]Unable to find 'app.py'. This command must be executed from the root of your Pipeless project[/red]")
+        sys.exit(1)
+
+    rprint(f"[yellow bold]Installing plugin with ID: {id}[/yellow bold]")
+    plugins_registry = get_plugins_registry()
+    install_success = False
+    for plugin in plugins_registry['plugins']:
+        if plugin.get("id") == id:
+            plugin_versions = plugin.get("versions")
+            if version is None:
+                # Default to latest version if no version specified
+                version = get_latest_plugin_version_number(plugin_versions)
+                rprint(f'[yellow]No version specified, installing latest version: {version}[/yellow]')
+
+            # Install the version
+            for plugin_version in plugin_versions:
+                if plugin_version.get('version') == version:
+                    repo_url = plugin_version.get("repo_url")
+                    tag_name = plugin_version.get("version")
+                    subdir = plugin_version.get("subdir")
+                    download_plugin_from_repo_tag(repo_url, tag_name, subdir, f'{plugins_root}/{id}')
+                    install_success = True
+                    python_deps = plugin_version.get("python_dependencies")
+                    install_pip_packages(python_deps)
+                    system_deps = plugin_version.get("system_dependencies")
+                    if len(system_deps) > 0:
+                        rprint(f"[yellow]The plugin {id} requires the following system dependencies, please install them now: {system_deps}[/yellow]")
+    if install_success:
+        rprint(f'[green]Plugin {id} sucessfully instaled![/green]')
+        rprint(f'[yellow]IMPORTANT: Remember to add "{id}" to the plugins execution order either in "config.yaml" or running:\n\t $ export PIPELESS_PLUGINS_ORDER=$PIPELESS_PLUGINS_ORDER,{id}[/yellow]')
+    else:
+        rprint('[red]The plugin (or plugin version) specified does not exit into the plugins registry[/red]')

cli/src/pipeless_ai_cli/commands/list.py

@@ -0,0 +1,22 @@
+import typer
+from rich import print as rprint
+
+from pipeless_ai_cli.lib.plugins import get_latest_plugin_version_dict, get_plugins_registry
+
+app = typer.Typer()
+
+@app.command(name="available-plugins", help="List plugins from the plugin registry that can be installed")
+def list_plugins():
+    plugins_registry = get_plugins_registry()
+    for plugin in plugins_registry['plugins']:
+        rprint(f'[green]{plugin.get("name")}[/green]')
+        rprint(f'\tID: {plugin.get("id")}')
+        rprint(f'\tDescription: {plugin.get("description")}')
+        version = get_latest_plugin_version_dict(plugin.get('versions'))
+        rprint(f'\tDocs URL: {version.get("docs_url")}')
+        rprint(f'\tLatest version: {version.get("version")}')
+        rprint(f'\tRepository URL: {version.get("repo_url")}')
+        rprint(f'\tRepository subdirectory: {version.get("subdir")}')
+        rprint(f'\tPython dependencies: {version.get("python_dependencies")}')
+        rprint(f'\tSystem dependencies: {version.get("system_dependencies")}')
+        rprint(f'\tPlugin dependencies: {version.get("plugin_dependencies")}')

cli/src/pipeless_ai_cli/lib/pip.py

@@ -0,0 +1,35 @@
+import subprocess
+import sys
+from rich import print as rprint
+from packaging.version import Version, parse
+from packaging.requirements import Requirement
+
+def install_pip_packages(packages: list[str]):
+    """
+    Receive a list of packages like ["package1@version", "package2@version"]
+    and installs the packages with pip
+    """
+    rprint("[yellow]Installing Python dependencies...[/yellow]")
+    for package in packages:
+        if "@" in package:
+            package_name, package_version = package.split('@')
+            version_specifier = Requirement(package_version)
+            version = parse(package_version)
+
+            # If the version specifier is "^", convert it to a compatible version specifier
+            if version_specifier.specs and version_specifier.specs[0][0] == "==":
+                compatible_version = version.base_version
+            else:
+                compatible_version = f">={version.base_version}"
+        else:
+            package_name = package
+            compatible_version = ""
+
+        pip_command = ["pip", "install", f"{package_name}{compatible_version}"]
+
+        try:
+            subprocess.check_call(pip_command)
+            rprint(f"\t[green]Successfully installed {package_name}{compatible_version}[/green]")
+        except subprocess.CalledProcessError:
+            rprint(f"\t[red]Failed to install pip package {package_name}{compatible_version}[/red]")
+            sys.exit(1)

cli/src/pipeless_ai_cli/lib/plugins.py

@@ -0,0 +1,64 @@
+import git
+import shutil
+import os
+import sys
+import json
+from rich import print as rprint
+import re
+
+def get_plugins_registry():
+    current_module_dir = os.path.dirname(__file__)
+    json_file_path = os.path.join(current_module_dir, "../plugins-registry.json")
+    try:
+        registry_json=open(json_file_path,"r")
+        registry_dict = json.load(registry_json)
+        registry_json.close()
+        return registry_dict
+    except Exception as e:
+        rprint(f"[red bold]An error occurred reading the registry:[/red bold] {e}")
+        sys.exit(1)
+
+def download_plugin_from_repo_tag(repo_url, tag_name, subdir, target_path):
+    download_repo_dir = "/tmp/temp_repo"
+    try:
+        if shutil.os.path.exists(download_repo_dir):
+            shutil.rmtree(download_repo_dir)
+        git.Repo.clone_from(repo_url, download_repo_dir)
+        repo = git.Repo(download_repo_dir)
+    except git.GitCommandError as e:
+        rprint(f'[red]Unable to download plugin repository "{repo_url}" into "{download_repo_dir}"[/red]')
+        print(e)
+        sys.exit(1)
+    try:
+        repo.git.checkout(tag_name)
+    except git.GitCommandError as e:
+        rprint(f'[red]The tag {tag_name} was not found on the target plugin repository[/red]')
+        shutil.rmtree(download_repo_dir) # Cleanup downloaded folders
+        print(e)
+        sys.exit(1)
+    source_path = os.path.join(download_repo_dir, subdir)
+    if shutil.os.path.exists(target_path):
+        shutil.rmtree(target_path)
+    shutil.copytree(source_path, target_path)
+    shutil.rmtree(download_repo_dir)
+
+def version_to_tuple(version_str):
+    """
+    Function to convert a version semver string to a tuple for comparison
+    Ex: "1.2.3" -> (1,2,3)
+    """
+    return tuple(map(int, re.findall(r'\d+', version_str)))
+
+def get_latest_plugin_version_number(plugin_versions: dict):
+    """
+    Takes the dict of versions of a plugin and returns the max version number
+    """
+    latest_version = get_latest_plugin_version_dict(plugin_versions)
+    return latest_version.get("version")
+
+def get_latest_plugin_version_dict(plugin_versions: dict):
+    """
+    Takes the dict of versions of a plugin and returns the max version dict
+    """
+    latest_version = next((v for v in plugin_versions if v["latest"]), None)
+    return latest_version

cli/src/pipeless_ai_cli/main.py

@@ -1,5 +1,5 @@
 import typer
-from .commands import create, run
+from .commands import create, install, run, list
 
 app = typer.Typer()
 
@@ -9,5 +9,8 @@
 def run_project(component: str = typer.Argument("all")):
     run.run_app(component)
 
+app.add_typer(install.app, name="install", help="Install project resources such as plugins")
+app.add_typer(list.app, name="list", help="List available project resources")
+
 if __name__ == "__main__":
     app()

cli/src/pipeless_ai_cli/plugins-registry.json

@@ -0,0 +1,21 @@
+{
+    "plugins": [
+        {
+            "name": "Pipeless Kafka Plugin",
+            "id": "kafka",
+            "description": "Allows to produce messages to a Kafka topic",
+            "versions": [
+                {
+                    "version": "plugins-kafka-0.1.0",
+                    "latest": true,
+                    "repo_url": "https://github.com/miguelaeh/pipeless.git",
+                    "docs_url": "https://pipeless.ai/docs/v0/plugins/kafka",
+                    "subdir": "plugins/kafka",
+                    "plugin_dependencies": [],
+                    "python_dependencies": ["confluent-kafka"],
+                    "system_dependencies": []
+                }
+            ]
+        }
+    ]
+}

core/src/pipeless_ai/lib/app/app.py

@@ -1,3 +1,4 @@
+import types
 from pipeless_ai.lib.timer import timer
 
 class PipelessApp():
@@ -6,7 +7,8 @@ class PipelessApp():
     """
 
     def __init__(self):
-        pass
+        self.plugins = types.SimpleNamespace()
+        self.__plugins_exec_graph = ()
 
     @timer
     def __before(self):