LocalStack

from localstack.cli.main import main

from localstack import config
from localstack.config import Directories

from localstack.aws.connect import connect_to

# Starting LocalStack programmatically
from localstack.cli.main import main
import sys

# Start LocalStack in Docker (equivalent to 'localstack start')
sys.argv = ['localstack', 'start', '--docker']
main()

# Using LocalStack's AWS client factory
from localstack.aws.connect import connect_to
import boto3

# Connect to LocalStack services using the factory
s3_client = connect_to().s3
dynamodb = connect_to().dynamodb

# Or use standard boto3 with LocalStack endpoint
s3 = boto3.client(
    's3',
    endpoint_url='http://localhost:4566',
    aws_access_key_id='test',
    aws_secret_access_key='test',
    region_name='us-east-1'
)

# Create a bucket and upload a file
s3.create_bucket(Bucket='my-test-bucket')
s3.put_object(Bucket='my-test-bucket', Key='test.txt', Body=b'Hello LocalStack!')

# List buckets to verify
response = s3.list_buckets()
print(f"Buckets: {[b['Name'] for b in response['Buckets']]}")

# Start LocalStack in Docker
localstack start

# Start with custom configuration
localstack start --docker -e DEBUG=1 -e SERVICES=s3,dynamodb

# Check service status
localstack status services

# View logs
localstack logs -f

# Stop LocalStack
localstack stop

# Entry point function
def main() -> None:
    """Main CLI entry point with profile and environment loading"""

# Core lifecycle commands  
localstack start [--docker|--host] [--detached] [options...]
localstack stop
localstack restart
localstack logs [-f] [-n N]
localstack wait [-t N]
localstack ssh

# Configuration functions
def load_environment(profiles: str = None, env=os.environ) -> list[str]:
    """Load configuration from profiles and environment variables"""

def parse_boolean_env(env_var_name: str) -> bool | None:
    """Parse boolean environment variables with LocalStack conventions"""

def is_env_true(env_var_name: str) -> bool:
    """Check if environment variable is set to true"""

# Directory management
class Directories:
    static_libs: str
    var_libs: str  
    cache: str
    tmp: str
    data: str
    config: str
    logs: str
    
    @classmethod
    def defaults() -> 'Directories':
        """Get default directory paths"""
        
    def mkdirs(self) -> None:
        """Create all directories"""

# Service management
class ServiceManager:
    def get_service(self, name: str) -> Service | None:
        """Get service instance by name"""
        
    def is_running(self, name: str) -> bool:
        """Check if service is running"""
        
    def get_states(self) -> dict[str, ServiceState]:
        """Get status of all services"""

# Service provider decorator
def aws_provider(
    api: str = None, 
    name: str = "default", 
    should_load: callable = None
) -> callable:
    """Decorator for registering AWS service providers"""

# Service states
class ServiceState(Enum):
    UNKNOWN = "unknown"
    AVAILABLE = "available"
    RUNNING = "running" 
    ERROR = "error"

# File utilities
def save_file(file_path: str, content: str, append: bool = False) -> None:
    """Save content to file"""

def load_file(file_path: str, default=None, mode: str = None) -> str:
    """Load file content with optional default"""

# Network utilities  
def get_free_tcp_port() -> int:
    """Get available TCP port"""

def is_port_open(port_or_url: str | int, http_path: str = None) -> bool:
    """Check if port is open"""

# Async utilities
async def run_sync(func: callable, *args, **kwargs):
    """Run synchronous function in async context"""

def run_async(coro, loop=None):
    """Run async coroutine in sync context"""

# CLI plugin system
class LocalstackCliPlugin:
    namespace: str = "localstack.plugins.cli"
    
    def attach(self, cli) -> None:
        """Attach commands to CLI"""

# Service plugins
class ServicePlugin:
    service: str
    api: str
    
    def create_service(self) -> Service:
        """Create service instance"""

# Plugin loading
def load_cli_plugins(cli) -> None:
    """Load all CLI plugins from namespace"""

class Service:
    """LocalStack service instance"""
    
    def __init__(
        self, 
        name: str, 
        start: callable = None, 
        check: callable = None, 
        skeleton=None, 
        active: bool = False, 
        stop: callable = None
    ): ...
    
    def start(self, asynchronous: bool = True) -> None:
        """Start the service"""
        
    def stop(self) -> None:
        """Stop the service"""
        
    def check(self, expect_shutdown: bool = False, print_error: bool = False) -> bool:
        """Check service health"""
        
    def is_enabled(self) -> bool:
        """Check if service is enabled"""

class ServiceState(Enum):
    """Service lifecycle states"""
    UNKNOWN = "unknown"
    AVAILABLE = "available"
    DISABLED = "disabled" 
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    STOPPED = "stopped"
    ERROR = "error"