tessl/pypi-localstack-core

The core library and runtime of LocalStack - a comprehensive AWS cloud emulator for local development and testing.

—

Pending

Overview

Eval results

Files

Plugins and Extensions

Name: tessl/pypi-localstack-core
Author: tessl

LocalStack provides an extensible plugin system that enables custom CLI commands, AWS service providers, runtime extensions, and package management. The plugin architecture allows for modular functionality while maintaining compatibility with LocalStack's core systems.

Capabilities

CLI Plugin System

Framework for extending the LocalStack CLI with custom commands and functionality.

class LocalstackCliPlugin:
    """
    Base class for LocalStack CLI plugins.
    
    CLI plugins can add custom commands, modify existing commands,
    and integrate with LocalStack's configuration system.
    
    Location: localstack.cli.plugin
    """
    
    namespace: str = "localstack.plugins.cli"
    
    def attach(self, cli) -> None:
        """
        Attach plugin commands and functionality to CLI.
        
        Called during CLI initialization to register custom commands,
        modify existing commands, or add global options.
        
        Args:
            cli: LocalStack CLI instance for command registration
        """

def load_cli_plugins(cli) -> None:
    """
    Discover and load all CLI plugins from the namespace.
    
    Automatically called during CLI initialization to load
    all plugins registered in the 'localstack.plugins.cli' namespace.
    
    Args:
        cli: LocalStack CLI instance
        
    Location: localstack.cli.plugin
    """

Service Plugin System

Plugin framework for implementing custom AWS service providers and extending existing services.

class ServicePlugin:
    """
    Plugin wrapper for AWS service provider implementations.
    
    Service plugins enable custom AWS service implementations
    that integrate with LocalStack's service management system.
    
    Location: localstack.services.plugins
    """
    
    service: str        # AWS service name (e.g., 'myservice')
    api: str           # AWS API identifier
    
    def create_service(self) -> 'Service':
        """
        Create service instance from this plugin.
        
        Returns:
            Configured Service instance with provider implementation
        """

def aws_provider(
    api: str = None, 
    name: str = "default", 
    should_load: callable = None
) -> callable:
    """
    Decorator for registering AWS service provider implementations.
    
    Use this decorator to mark classes as AWS service providers
    that implement specific AWS APIs.
    
    Args:
        api: AWS service API name (e.g., 's3', 'lambda', 'myservice')
        name: Provider name for multiple implementations (default: "default")
        should_load: Optional function returning bool to control loading
        
    Returns:
        Decorator function for provider classes
        
    Location: localstack.services.plugins
    """

class ServiceProvider(Protocol):
    """
    Protocol interface for AWS service provider implementations.
    
    Service providers implement AWS service operations and integrate
    with LocalStack's AWS Service Framework (ASF).
    
    Location: localstack.services.plugins
    """
    
    service: str  # AWS service name this provider implements

# Plugin namespace for AWS service providers
PLUGIN_NAMESPACE: str = "localstack.aws.provider"

Extension System

Runtime extension loading and management for enhanced LocalStack functionality.

class Extension:
    """
    Base class for LocalStack runtime extensions.
    
    Extensions can modify LocalStack behavior, add new capabilities,
    and integrate with the service lifecycle.
    
    Location: localstack.extensions
    """
    
    name: str           # Extension name
    version: str        # Extension version
    
    def on_extension_load(self) -> None:
        """Called when extension is loaded during startup"""
        
    def on_platform_start(self) -> None:
        """Called when LocalStack platform starts"""
        
    def on_platform_shutdown(self) -> None:
        """Called when LocalStack platform shuts down"""

def load_extensions() -> list['Extension']:
    """
    Discover and load all registered extensions.
    
    Returns:
        List of loaded extension instances
        
    Location: localstack.extensions
    """

def register_extension(extension_class: type) -> None:
    """
    Register extension class for automatic loading.
    
    Args:
        extension_class: Extension class to register
        
    Location: localstack.extensions
    """

Package Management System

Package plugin management for third-party integrations and additional services.

class PackagePlugin:
    """
    Plugin for managing third-party packages and integrations.
    
    Package plugins enable installation and management of
    external tools, services, and dependencies.
    
    Location: localstack.packages
    """
    
    name: str               # Package name
    version: str            # Package version
    dependencies: list[str] # Package dependencies
    
    def install(self, target_dir: str) -> bool:
        """
        Install package to target directory.
        
        Args:
            target_dir: Installation target directory
            
        Returns:
            True if installation successful
        """
        
    def is_installed(self) -> bool:
        """
        Check if package is already installed.
        
        Returns:
            True if package is available
        """
        
    def get_version(self) -> str:
        """
        Get installed package version.
        
        Returns:
            Version string or None if not installed
        """

def install_package(
    package_name: str, 
    version: str = None,
    target: str = None
) -> bool:
    """
    Install package using package manager.
    
    Args:
        package_name: Name of package to install
        version: Specific version to install (None for latest)
        target: Installation target directory
        
    Returns:
        True if installation successful
        
    Location: localstack.packages
    """

def list_available_packages() -> list[dict[str, str]]:
    """
    List packages available for installation.
    
    Returns:
        List of package information dictionaries
        
    Location: localstack.packages
    """

Plugin Development

Creating CLI Plugins

Example implementation of a custom CLI plugin:

from localstack.cli.plugin import LocalstackCliPlugin
import click

class MyCliPlugin(LocalstackCliPlugin):
    """Custom CLI plugin example"""
    
    def attach(self, cli):
        """Add custom commands to CLI"""
        
        @cli.group()
        def mycommands():
            """Custom command group"""
            pass
            
        @mycommands.command()
        @click.option('--target', help='Target for operation')
        def deploy(target):
            """Deploy custom resources"""
            click.echo(f"Deploying to {target}")
            # Custom deployment logic
            
        @mycommands.command() 
        def status():
            """Check custom resource status"""
            click.echo("Checking status...")
            # Custom status logic

# Register plugin via entry point in setup.py/pyproject.toml:
# [project.entry-points."localstack.plugins.cli"]
# my-plugin = "mypackage.cli:MyCliPlugin"

Creating Service Providers

Example implementation of a custom AWS service provider:

from localstack.services.plugins import aws_provider, ServiceProvider
from localstack.aws.api import RequestContext
from typing import Dict, Any

@aws_provider(api="myservice")
class MyServiceProvider(ServiceProvider):
    """Custom AWS service provider"""
    
    service = "myservice"
    
    def create_resource(
        self, 
        context: RequestContext, 
        request: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Create a custom resource.
        
        Args:
            context: Request context with AWS metadata
            request: Parsed request parameters
            
        Returns:
            AWS-compatible response dictionary
        """
        resource_name = request.get("ResourceName")
        resource_id = f"resource-{self._generate_id()}"
        
        # Store resource (implementation specific)
        self._store_resource(resource_id, {
            "ResourceId": resource_id,
            "ResourceName": resource_name,
            "Status": "ACTIVE"
        })
        
        return {
            "ResourceId": resource_id,
            "ResourceArn": f"arn:aws:myservice:us-east-1:000000000000:resource/{resource_id}"
        }
    
    def list_resources(
        self, 
        context: RequestContext, 
        request: Dict[str, Any]
    ) -> Dict[str, Any]:
        """List all resources"""
        resources = self._list_stored_resources()
        
        return {
            "Resources": [
                {
                    "ResourceId": r["ResourceId"],
                    "ResourceName": r["ResourceName"], 
                    "Status": r["Status"]
                }
                for r in resources
            ]
        }
    
    def _generate_id(self) -> str:
        """Generate unique resource ID"""
        import uuid
        return str(uuid.uuid4())[:8]
        
    def _store_resource(self, resource_id: str, resource: Dict[str, Any]) -> None:
        """Store resource (implement with your preferred storage)"""
        pass
        
    def _list_stored_resources(self) -> List[Dict[str, Any]]:
        """List stored resources"""
        return []

# Register via entry point:
# [project.entry-points."localstack.aws.provider"]  
# myservice = "mypackage.services:MyServiceProvider"

Creating Extensions

Example runtime extension that adds custom functionality:

from localstack.extensions import Extension
import logging

LOG = logging.getLogger(__name__)

class MyExtension(Extension):
    """Custom LocalStack extension"""
    
    name = "my-extension"
    version = "1.0.0"
    
    def on_extension_load(self):
        """Initialize extension on load"""
        LOG.info("Loading my custom extension")
        self._setup_custom_functionality()
        
    def on_platform_start(self):
        """Execute when LocalStack starts"""
        LOG.info("LocalStack platform starting - extension active")
        self._start_background_tasks()
        
    def on_platform_shutdown(self):
        """Cleanup when LocalStack stops"""
        LOG.info("LocalStack platform shutting down")
        self._cleanup_resources()
        
    def _setup_custom_functionality(self):
        """Setup extension-specific functionality"""
        # Add custom middleware, modify configs, etc.
        pass
        
    def _start_background_tasks(self):
        """Start extension background tasks"""
        # Start monitoring, periodic tasks, etc.
        pass
        
    def _cleanup_resources(self):
        """Clean up extension resources"""
        # Stop tasks, close connections, etc.
        pass

# Register extension
from localstack.extensions import register_extension
register_extension(MyExtension)

Creating Package Plugins

Example package plugin for managing external dependencies:

from localstack.packages import PackagePlugin
import subprocess
import os

class RedisPackage(PackagePlugin):
    """Redis server package plugin"""
    
    name = "redis"
    version = "7.0.0"
    dependencies = []
    
    def install(self, target_dir: str) -> bool:
        """Install Redis server"""
        try:
            # Download and install Redis
            redis_url = f"https://download.redis.io/releases/redis-{self.version}.tar.gz"
            
            # Create target directory
            os.makedirs(target_dir, exist_ok=True)
            
            # Download, extract, and build Redis
            subprocess.run([
                "curl", "-L", redis_url, "|", 
                "tar", "xz", "-C", target_dir
            ], check=True)
            
            redis_dir = os.path.join(target_dir, f"redis-{self.version}")
            subprocess.run(["make"], cwd=redis_dir, check=True)
            
            return True
        except Exception as e:
            print(f"Redis installation failed: {e}")
            return False
            
    def is_installed(self) -> bool:
        """Check if Redis is installed"""
        try:
            subprocess.run(["redis-server", "--version"], 
                         check=True, capture_output=True)
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            return False
            
    def get_version(self) -> str:
        """Get installed Redis version"""
        try:
            result = subprocess.run(
                ["redis-server", "--version"],
                capture_output=True, text=True, check=True
            )
            # Parse version from output
            return result.stdout.split()[2]
        except Exception:
            return None

# Register via entry point:
# [project.entry-points."localstack.packages"]
# redis = "mypackage.packages:RedisPackage"

Plugin Discovery and Loading

Entry Point Configuration

Plugins are discovered via Python entry points in pyproject.toml:

[project.entry-points."localstack.plugins.cli"]
my-cli-plugin = "mypackage.cli:MyCliPlugin"

[project.entry-points."localstack.aws.provider"] 
myservice = "mypackage.services:MyServiceProvider"
custom-s3 = "mypackage.s3:CustomS3Provider"

[project.entry-points."localstack.extensions"]
my-extension = "mypackage.extensions:MyExtension"

[project.entry-points."localstack.packages"]
redis = "mypackage.packages:RedisPackage"
elasticsearch = "mypackage.packages:ElasticsearchPackage"

Plugin Loading Process

def discover_plugins(namespace: str) -> list[object]:
    """
    Discover plugins from entry point namespace.
    
    Args:
        namespace: Entry point namespace to search
        
    Returns:
        List of plugin instances
    """

def load_plugin(plugin_class: type, config: dict = None) -> object:
    """
    Load and initialize plugin instance.
    
    Args:
        plugin_class: Plugin class to instantiate
        config: Optional plugin configuration
        
    Returns:
        Initialized plugin instance
    """

Plugin Configuration

Plugins can access LocalStack configuration and define their own settings:

class ConfigurablePlugin(LocalstackCliPlugin):
    """Plugin with configuration support"""
    
    def __init__(self):
        self.config = self._load_config()
    
    def _load_config(self) -> dict:
        """Load plugin-specific configuration"""
        from localstack.config import config
        
        return {
            "enabled": config.get("MY_PLUGIN_ENABLED", "true").lower() == "true",
            "api_key": config.get("MY_PLUGIN_API_KEY"),
            "endpoint": config.get("MY_PLUGIN_ENDPOINT", "https://api.example.com")
        }
    
    def attach(self, cli):
        if not self.config["enabled"]:
            return  # Skip if disabled
            
        @cli.command()
        def my_command():
            """Custom command with configuration"""
            api_key = self.config["api_key"]
            endpoint = self.config["endpoint"]
            # Use configuration in command logic

Plugin Distribution

Package Structure

Recommended structure for plugin packages:

my-localstack-plugin/
├── pyproject.toml              # Package configuration with entry points
├── README.md                   # Plugin documentation
├── my_plugin/
│   ├── __init__.py
│   ├── cli.py                  # CLI plugin implementation
│   ├── services.py             # Service provider implementations  
│   ├── extensions.py           # Runtime extensions
│   └── packages.py             # Package managers
└── tests/
    ├── test_cli.py
    ├── test_services.py
    └── test_extensions.py

Installation and Usage

Users install plugins as regular Python packages:

# Install plugin
pip install my-localstack-plugin

# Plugin commands are automatically available
localstack my-custom-command --help

# Plugin services are automatically loaded
localstack start  # Custom services included

# Plugin packages available via LPM
localstack lpm install my-custom-package

Plugin Testing

Test plugins using LocalStack's testing framework:

import pytest
from localstack.testing.pytest import localstack

def test_my_plugin_cli():
    """Test custom CLI command"""
    from click.testing import CliRunner
    from my_plugin.cli import MyCliPlugin
    
    runner = CliRunner()
    # Test CLI plugin functionality

@pytest.mark.aws  
def test_my_service_provider(localstack):
    """Test custom service provider"""
    import boto3
    
    # Create client for custom service
    client = boto3.client(
        'myservice',
        endpoint_url='http://localhost:4566',
        aws_access_key_id='test',
        aws_secret_access_key='test'
    )
    
    # Test service operations
    response = client.create_resource(ResourceName='test')
    assert 'ResourceId' in response

Install with Tessl CLI