Skip to content

Plugin Development Guide

This guide covers how to create plugins for Hop3: builders, language toolchains, deployers, addons, proxies, and operating system support.

Overview

Hop3 uses a plugin-based architecture built on pluggy, the same plugin framework used by pytest. Plugins allow you to extend Hop3 with:

  • Builders: Orchestrate builds (local vs containerized), e.g., LocalBuilder, DockerBuilder
  • Language Toolchains: Language-specific build tooling, e.g., Python, Node, Ruby, Go, Rust
  • Deployers: Run build artifacts, e.g., uWSGI, Docker Compose, Static
  • Addons: Backing services (databases, caches), e.g., PostgreSQL, Redis, MySQL
  • Proxies: Configure reverse proxies, e.g., Nginx, Caddy, Traefik
  • OS Implementations: Support Linux distributions, e.g., Debian family, Red Hat family

When to Create a Plugin

Create a plugin when you want to:

  • Add support for a new programming language (language toolchain)
  • Support a new build orchestration method (builder)
  • Support a new deployment runtime (deployer)
  • Integrate a new backing service like MongoDB, Elasticsearch (addon)
  • Add support for a new reverse proxy (proxy)
  • Support a new Linux distribution (OS implementation)

Modify core code when you need to:

  • Change the plugin system itself
  • Modify the deployment pipeline orchestration
  • Change database models or API contracts
  • Update the CLI or web UI

Plugin Types

1. Builders

Builders orchestrate the build process. There are two levels: - Level 1 (Orchestrators): Decide how to build (local vs containerized) - Level 2 (Language Toolchains): Language-specific build logic (used by LocalBuilder)

Protocol: Builder (from hop3.core.protocols)

Required attributes: - name (str): Unique identifier (e.g., "python", "docker") - context (DeploymentContext): Deployment context with app info

Required methods: - accept() -> bool: Return True if this strategy can build the app - build() -> BuildArtifact: Execute build and return artifact info

Example (simplified Python builder):

from pathlib import Path
from hop3.core.protocols import Builder, BuildArtifact, DeploymentContext


class PythonBuildStrategy:
    """Build Python applications using virtualenv."""

    name = "python"

    def __init__(self, context: DeploymentContext):
        self.context = context

    def accept(self) -> bool:
        """Accept if requirements.txt or pyproject.toml exists."""
        src_path = self.context.source_path
        return (src_path / "requirements.txt").exists() or
            (src_path / "pyproject.toml").exists()

    def build(self) -> BuildArtifact:
        """Create virtualenv and install dependencies."""
        app_name = self.context.app_name
        venv_path = Path(f"/home/hop3/apps/{app_name}/venv")

        # Create virtualenv
        subprocess.run(["python3", "-m", "venv", str(venv_path)], check=True)

        # Install dependencies
        pip = venv_path / "bin" / "pip"
        subprocess.run([str(pip), "install", "-r", "requirements.txt"],
                       cwd=self.context.source_path, check=True)

        return BuildArtifact(
            kind="virtualenv",
            location=str(venv_path),
            metadata={"python_version": "3.11"}
        )

2. Language Toolchains

Language toolchains provide language-specific build logic. They are used by LocalBuilder to build applications for specific programming languages.

Protocol: LanguageToolchain (from hop3.core.protocols)

Required methods: - accept() -> bool: Return True if this toolchain can build the app - build() -> BuildArtifact: Execute build and return artifact info

Hook: get_language_toolchains()

Location: hop3/plugins/build/language_toolchains/

Available toolchains: Python, Node.js, Ruby, Go, Rust, Java, PHP, Clojure, .NET, Elixir, Static

3. Deployers

Deployers run build artifacts and manage their lifecycle.

Protocol: Deployer (from hop3.core.protocols)

Required attributes: - name (str): Unique identifier (e.g., "uwsgi", "docker-compose") - context (DeploymentContext): Deployment context - artifact (BuildArtifact): Build artifact to deploy

Required methods: - accept() -> bool: Return True if this strategy can deploy the artifact - deploy(deltas: dict | None = None) -> DeploymentInfo: Deploy the artifact - stop() -> None: Stop the running application - check_status() -> bool: Check if app is actually running - scale(deltas: dict[str, int] | None = None) -> None: Scale workers

Example (simplified Docker Compose deployer):

import subprocess
from hop3.core.protocols import Deployer, DeploymentInfo

class DockerComposeDeployer:
    """Deploy applications using Docker Compose."""

    name = "docker-compose"

    def __init__(self, context, artifact):
        self.context = context
        self.artifact = artifact

    def accept(self) -> bool:
        """Accept if artifact is a docker-image."""
        return self.artifact.kind == "docker-image"

    def deploy(self, deltas=None) -> DeploymentInfo:
        """Run docker-compose up."""
        src_path = self.context.source_path

        env = {"HOP3_IMAGE_TAG": self.artifact.location}
        subprocess.run(
            ["docker", "compose", "up", "-d", "--remove-orphans"],
            cwd=src_path,
            check=True,
            env=env
        )

        return DeploymentInfo(
            protocol="http",
            address="127.0.0.1",
            port=8080
        )

    def stop(self):
        """Run docker-compose down."""
        src_path = self.context.source_path
        subprocess.run(["docker", "compose", "down"], cwd=src_path)

    def check_status(self) -> bool:
        """Check if containers are running."""
        result = subprocess.run(
            ["docker", "compose", "ps", "--format", "{{.State}}"],
            cwd=self.context.source_path,
            capture_output=True,
            text=True
        )
        return "running" in result.stdout.lower()

    def scale(self, deltas=None):
        """Scale services."""
        if not deltas:
            return

        scale_args = []
        for service, count in deltas.items():
            scale_args.extend(["--scale", f"{service}={count}"])

        cmd = ["docker", "compose", "up", "-d"] + scale_args
        subprocess.run(cmd, cwd=self.context.source_path, check=True)

4. Addons

Addons manage backing services (databases, caches, etc.). They are independent resources that can be shared across applications.

Protocol: Addon (from hop3.core.protocols)

Hook: get_addons()

Location: hop3/plugins/{postgresql,mysql,redis}/

Required attributes: - name (str): Addon type (e.g., "postgres", "redis") - addon_name (str): Specific instance name

Required methods: - create() -> None: Create the service instance - destroy() -> None: Destroy the service instance - get_connection_details() -> dict[str, str]: Return environment variables for connection - backup() -> Path: Create a backup - restore(backup_path: Path) -> None: Restore from backup - info() -> dict: Get service information

Example (simplified Redis service):

from pathlib import Path
from hop3.core.protocols import Addon


class RedisService:
    """Manage Redis service instances."""

    name = "redis"

    def __init__(self, *, addon_name: str):
        self.addon_name = addon_name

    def create(self):
        """Create Redis instance."""
        port = self._allocate_port()
        config_path = Path(f"/home/hop3/services/redis/{self.addon_name}.conf")

        # Write Redis config
        config_path.parent.mkdir(parents=True, exist_ok=True)
        config_path.write_text(f"port {port}\nbind 127.0.0.1\n")

        # Start Redis with systemd
        subprocess.run([
            "systemctl", "start", f"redis-{self.addon_name}"
        ], check=True)

    def destroy(self):
        """Destroy Redis instance."""
        subprocess.run([
            "systemctl", "stop", f"redis-{self.addon_name}"
        ])

    def get_connection_details(self) -> dict[str, str]:
        """Return connection environment variables."""
        port = self._get_port()
        return {
            "REDIS_URL": f"redis://127.0.0.1:{port}/0"
        }

    def backup(self) -> Path:
        """Backup Redis data."""
        # Trigger BGSAVE and copy RDB file
        subprocess.run(["redis-cli", "BGSAVE"])
        # ... implementation details
        return Path(f"/backups/redis-{self.addon_name}.rdb")

    def restore(self, backup_path: Path):
        """Restore from backup."""
        # Stop Redis, copy RDB file, restart
        # ... implementation details
        pass

    def info(self) -> dict:
        """Get service info."""
        return {
            "status": "running",
            "version": "7.0",
            "port": self._get_port()
        }

5. Proxies

Proxies configure reverse proxies for applications.

Base Class: BaseProxy (from hop3.core.protocols)

Hook: get_proxies()

Location: hop3/plugins/proxy/{nginx,caddy,traefik}/

Required attributes: - app (App): Application instance - env (Env): Environment variables - workers (dict[str, str]): Worker configurations

Required methods (abstract): - get_proxy_name() -> str: Return proxy name ("nginx", "caddy", "traefik") - setup_backend() -> None: Configure backend connection - setup_certificates() -> None: Setup SSL certificates - setup_cache() -> None: Configure caching - setup_static() -> None: Configure static file serving - extra_setup() -> None: Additional proxy-specific setup - generate_config() -> None: Generate proxy config file - check_config() -> None: Validate config - reload_proxy() -> None: Reload proxy to apply changes

Capabilities (class attributes; override the BaseProxy defaults to match what the proxy can do): - backend_protocols: frozenset[str]: what it can speak to an app's port (default {"http"}; add "fastcgi" for php-fpm pools) - serves_files: bool: whether it serves files from disk, including the ACME challenge webroot (default True) - fronts_waf: bool: whether it can route an app through its WAF (default False) - hosts_certbot: bool: whether Hop3's certbot issuance and renewal work behind it (default False)

A deploy refuses an app that needs a capability the proxy does not declare, so declare only what the proxy does.

The setup() method is already implemented in BaseProxy and orchestrates the setup process.

Example (simplified Nginx proxy):

from dataclasses import dataclass
from hop3.core.protocols import BaseProxy

@dataclass(frozen=True)
class NginxVirtualHost(BaseProxy):
    """Nginx reverse proxy configuration."""

    def get_proxy_name(self) -> str:
        return "nginx"

    def setup_backend(self):
        """Configure backend connection."""
        # Use Unix socket or TCP
        if self.env.get("NGINX_SOCKET"):
            socket_path = f"/tmp/{self.app_name}.sock"
            self.update_env("NGINX_SOCKET", socket_path)
        else:
            port = self.app.port
            self.update_env("NGINX_BACKEND", f"127.0.0.1:{port}")

    def setup_certificates(self):
        """Setup SSL certificates."""
        # Generate self-signed cert or use Let's Encrypt
        pass

    def setup_cache(self):
        """Configure caching."""
        if self.env.get("NGINX_CACHE_ENABLE"):
            self.update_env("NGINX_CACHE_PATH", f"/var/cache/nginx/{self.app_name}")

    def setup_static(self):
        """Configure static files."""
        # Use BaseProxy's get_static_paths() helper
        static_paths = self.get_static_paths()
        # ... generate location blocks

    def extra_setup(self):
        """Additional Nginx-specific setup."""
        pass

    def generate_config(self):
        """Generate Nginx config file."""
        template = self._load_template("nginx.conf.j2")
        config = template.render(env=self.env, app=self.app)

        config_path = Path(f"/etc/nginx/sites-available/{self.app_name}.conf")
        config_path.write_text(config)

    def check_config(self):
        """Validate Nginx config."""
        subprocess.run(["nginx", "-t"], check=True)

    def reload_proxy(self):
        """Reload Nginx."""
        subprocess.run(["systemctl", "reload", "nginx"], check=True)

6. OS Implementations

OS implementations handle operating system-specific configuration.

Protocol: OS (from hop3.core.protocols)

Hook: get_os_implementations()

Location: hop3/plugins/oses/

Required attributes: - name (str): OS identifier (e.g., "debian12", "ubuntu2204") - display_name (str): Human-readable name - packages (list[str]): Required system packages

Required methods: - detect() -> bool: Check if this OS matches the current system - setup_server() -> None: Install dependencies and configure system - ensure_packages(packages, update=True) -> None: Install system packages - ensure_user(user, home, shell, group) -> None: Create system user

Example (simplified Debian family support):

from pathlib import Path
from hop3.core.protocols import OS

class DebianFamilyOS:
    """Support for Debian-based distributions."""

    name = "debian"
    display_name = "Debian Family (Debian, Ubuntu, etc.)"
    packages = [
        "python3", "python3-pip", "python3-venv",
        "nginx", "git", "build-essential"
    ]

    def detect(self) -> bool:
        """Check if this is a Debian-based OS."""
        os_release = Path("/etc/os-release").read_text()
        return "debian" in os_release.lower() or "ubuntu" in os_release.lower()

    def setup_server(self):
        """Setup server for hop3."""
        # Update package lists
        subprocess.run(["apt-get", "update"], check=True)

        # Create hop3 user
        self.ensure_user("hop3", "/home/hop3", "/bin/bash", "hop3")

        # Install required packages
        self.ensure_packages(self.packages)

    def ensure_packages(self, packages, update=True):
        """Install packages with apt."""
        if update:
            subprocess.run(["apt-get", "update"])

        subprocess.run(
            ["apt-get", "install", "-y"] + packages,
            check=True
        )

    def ensure_user(self, user, home, shell, group):
        """Create user if not exists."""
        try:
            subprocess.run(["id", user], check=True, capture_output=True)
        except subprocess.CalledProcessError:
            # User doesn't exist, create it
            subprocess.run([
                "useradd", "-m",
                "-d", home,
                "-s", shell,
                "-U", user  # Create group with same name
            ], check=True)

Registering Plugins

Internal Plugins (in hop3.plugins)

Internal plugins are automatically discovered by scanning the hop3.plugins package.

Plugin structure: 

hop3/plugins/
├── my_plugin/
│   ├── __init__.py      # Must import plugin
│   ├── plugin.py        # Plugin class with hooks
│   ├── builder.py       # Strategy implementations
│   └── deployer.py

Plugin class (plugin.py): 

from hop3.core.hooks import hookimpl
from .toolchain import MyLanguageToolchain
from .deployer import MyDeployer

class MyPlugin:
    """My custom plugin."""

    name = "my_plugin"

    @hookimpl
    def get_language_toolchains(self) -> list:
        """Return language toolchain classes."""
        return [MyLanguageToolchain]

    @hookimpl
    def get_deployers(self) -> list:
        """Return deployer classes."""
        return [MyDeployer]

# Auto-register when module is imported
plugin = MyPlugin()

Available hooks: - get_builders() - Return Builder classes - get_language_toolchains() - Return LanguageToolchain classes - get_deployers() - Return Deployer classes - get_addons() - Return Addon classes - get_proxies() - Return Proxy classes - get_os_implementations() - Return OS classes - get_di_providers() - Return Dishka Provider instances - cli_commands() - Register CLI commands

Important: The __init__.py must import the plugin: 

from .plugin import plugin

__all__ = ["plugin"]

External Plugins (via setuptools entry points)

External plugins are distributed as separate Python packages and registered via setuptools entry points.

Entry point in setup.py or pyproject.toml:

# pyproject.toml
[project.entry-points."hop3.plugins"]
my_plugin = "my_hop3_plugin.plugin:plugin"

Or in setup.py: 

setup(
    name="my-hop3-plugin",
    entry_points={
        "hop3.plugins": [
            "my_plugin = my_hop3_plugin.plugin:plugin"
        ]
    }
)

Package structure: 

my-hop3-plugin/
├── pyproject.toml
├── src/
│   └── my_hop3_plugin/
│       ├── __init__.py
│       ├── plugin.py
│       ├── builder.py
│       └── deployer.py

Plugin class: 

# src/my_hop3_plugin/plugin.py
from hop3.core.hooks import hookimpl
from .builder import MyBuilder

class MyExternalPlugin:
    name = "my_external_plugin"

    @hookimpl
    def get_builders(self) -> list:
        return [MyBuilder]

plugin = MyExternalPlugin()

Testing Plugins

Unit Testing

Test your strategy classes in isolation:

# tests/test_my_builder.py
from pathlib import Path
from hop3.core.protocols import DeploymentContext
from my_hop3_plugin.builder import MyBuilder

def test_accept():
    """Test that builder accepts correct apps."""
    context = DeploymentContext(
        app_name="test",
        source_path=Path("/tmp/test"),
        app_config={}
    )

    builder = MyBuilder(context)
    assert builder.accept() is True

def test_build():
    """Test build process."""
    # ... create test fixtures
    builder = MyBuilder(context)
    artifact = builder.build()

    assert artifact.kind == "my-artifact-type"
    assert Path(artifact.location).exists()

Integration Testing

These tests cover plugin registration and discovery:

# tests/test_plugin_registration.py
from hop3.core.plugins import get_plugin_manager

def test_plugin_registered():
    """Test that plugin is discovered."""
    pm = get_plugin_manager()
    strategies = pm.hook.get_builders()
    strategy_classes = [cls for sublist in strategies for cls in sublist]

    names = [getattr(cls, "name", None) for cls in strategy_classes]
    assert "my_builder" in names

System Testing

Test the full deployment pipeline with your plugin:

# tests/test_deployment.py
def test_deploy_with_my_plugin(tmp_path):
    """Test full deployment using my plugin."""
    # Create test app
    app = App(name="test", runtime="my-runtime")

    # Deploy app
    do_deploy(app)

    # Verify deployment
    assert app.is_running

Best Practices

1. Strategy Naming

  • Use lowercase names: "python", not "Python"
  • Be specific: prefer "docker-compose" over the generic "docker"
  • Avoid conflicts with existing strategies

2. Error Handling

  • Raise Abort from hop3.lib for user-facing errors
  • Use log() from hop3.lib for informational messages
  • Provide helpful error messages with available options
from hop3.lib import Abort, log

def deploy(self, deltas=None):
    try:
        # ... deployment logic
        log("Deployment successful", fg="green")
    except FileNotFoundError:
        msg = "Docker not found. Please install Docker first."
        raise Abort(msg)

3. Idempotency

Make operations safe to run multiple times:

def create(self):
    """Create service (idempotent)."""
    if self._already_exists():
        log(f"Service {self.addon_name} already exists", fg="yellow")
        return

    # ... create service

4. Resource Cleanup

Always clean up resources in destroy() methods:

def destroy(self):
    """Clean up all service resources."""
    # Stop service
    self.stop()

    # Remove config files
    self._remove_config()

    # Remove data (if appropriate)
    if self.env.get("REMOVE_DATA"):
        self._remove_data()

5. Configuration

Use environment variables for configuration:

def accept(self) -> bool:
    """Check if builder is enabled."""
    # Allow disabling builder via env var
    if self.context.app_config.get("DISABLE_MY_BUILDER"):
        return False

    return self._detect_app_type()

6. Documentation

Document your strategies with clear docstrings:

class MyBuilder:
    """Build strategy for MyFramework applications.

    This builder supports MyFramework 2.0+ applications with
    either requirements.txt or Pipfile dependencies.

    Environment Variables:
        MY_BUILDER_VERSION: Specify builder version (default: latest)
        MY_BUILDER_CACHE: Enable build caching (default: true)

    Example hop3.toml:
        [build]
        builder = "my-builder"

        [build.env]
        MY_BUILDER_VERSION = "2.1"
    """

Common Pitfalls

1. Forgetting to Register Plugin Instance

Wrong: 

# plugin.py
class MyPlugin:
    @hookimpl
    def get_builders(self):
        return [MyBuilder]

# Missing: plugin = MyPlugin()

Correct: 

class MyPlugin:
    @hookimpl
    def get_builders(self):
        return [MyBuilder]

# Must create instance for auto-registration
plugin = MyPlugin()

2. Not Importing Plugin in __init__.py

Wrong: 

# __init__.py
# Empty or missing import

Correct: 

# __init__.py
from .plugin import plugin

__all__ = ["plugin"]

3. Returning Strategy Instances Instead of Classes

Wrong: 

@hookimpl
def get_builders(self):
    # Don't instantiate here
    return [MyBuilder()]

Correct: 

@hookimpl
def get_builders(self):
    # Return classes, not instances
    return [MyBuilder]

4. Hardcoding Paths

Wrong: 

def build(self):
    venv_path = "/home/hop3/venv"  # Hardcoded

Correct: 

def build(self):
    from hop3.config import HopConfig
    cfg = HopConfig.get_instance()
    venv_path = cfg.APP_ROOT / self.context.app_name / "venv"

5. Not Checking check_status() Properly

Wrong: 

def check_status(self) -> bool:
    # Just check if config exists (unreliable)
    return Path(f"/etc/systemd/system/{self.app}.service").exists()

Correct: 

def check_status(self) -> bool:
    # Actually verify process is running
    result = subprocess.run(
        ["systemctl", "is-active", f"{self.app}.service"],
        capture_output=True
    )
    return result.returncode == 0

Next Steps

Getting Help

  • Review existing plugins in hop3/plugins/
  • Check hook specifications in hop3/core/hookspecs.py
  • Check protocol definitions in hop3/core/protocols.py
  • Ask questions in the project discussions
  • Read the pluggy documentation: https://pluggy.readthedocs.io/