AWS Services

class ServiceManager:
    """
    Central manager for LocalStack service instances and lifecycle.
    
    Location: localstack.services.plugins
    """
    
    def get_service(self, name: str) -> Service | None:
        """
        Retrieve service instance by name.
        
        Args:
            name: AWS service name (e.g., 's3', 'lambda', 'dynamodb')
            
        Returns:
            Service instance if found, None otherwise
        """
        
    def add_service(self, service: 'Service') -> bool:
        """
        Register a new service with the manager.
        
        Args:
            service: Service instance to register
            
        Returns:
            True if service was added successfully
        """
        
    def require(self, name: str) -> 'Service':
        """
        Get service instance, raising exception if not found.
        
        Args:
            name: AWS service name
            
        Returns:
            Service instance
            
        Raises:
            ServiceException: If service not found or not available
        """
        
    def is_running(self, name: str) -> bool:
        """
        Check if service is currently running.
        
        Args:
            name: AWS service name
            
        Returns:
            True if service is in RUNNING state
        """
        
    def get_states(self) -> dict[str, 'ServiceState']:
        """
        Get current state of all registered services.
        
        Returns:
            Dictionary mapping service names to their current states
        """

class Service:
    """
    Individual AWS service instance with lifecycle management.
    
    Location: localstack.services.plugins
    """
    
    def __init__(
        self,
        name: str,
        start: callable = None,
        check: callable = None, 
        skeleton=None,
        active: bool = False,
        stop: callable = None
    ):
        """
        Initialize service instance.
        
        Args:
            name: Service name (e.g., 's3', 'lambda')
            start: Function to start the service
            check: Function to check service health
            skeleton: ASF skeleton for API handling
            active: Whether service starts automatically
            stop: Function to stop the service
        """
    
    def start(self, asynchronous: bool = True) -> None:
        """
        Start the service instance.
        
        Args:
            asynchronous: Whether to start asynchronously
        """
        
    def stop(self) -> None:
        """Stop the service instance and clean up resources."""
        
    def check(
        self, 
        expect_shutdown: bool = False, 
        print_error: bool = False
    ) -> bool:
        """
        Check service health and readiness.
        
        Args:
            expect_shutdown: Whether service shutdown is expected
            print_error: Whether to print error details
            
        Returns:
            True if service is healthy
        """
        
    def is_enabled(self) -> bool:
        """
        Check if service is enabled in current configuration.
        
        Returns:
            True if service is enabled
        """

class ServiceState(Enum):
    """
    Service lifecycle states.
    
    Location: localstack.services.plugins
    """
    UNKNOWN = "unknown"         # State cannot be determined
    AVAILABLE = "available"     # Service available but not started
    DISABLED = "disabled"       # Service disabled in configuration
    STARTING = "starting"       # Service is starting up
    RUNNING = "running"         # Service is fully operational  
    STOPPING = "stopping"       # Service is shutting down
    STOPPED = "stopped"         # Service has been stopped
    ERROR = "error"            # Service encountered an error

def aws_provider(
    api: str = None, 
    name: str = "default", 
    should_load: callable = None
) -> callable:
    """
    Decorator for registering AWS service providers.
    
    Args:
        api: AWS service API name (e.g., 's3', 'lambda')
        name: Provider name (default: "default")
        should_load: Optional function to determine if provider should load
        
    Returns:
        Decorator function for provider classes
        
    Location: localstack.services.plugins
    """

class ServiceProvider(Protocol):
    """
    Protocol for AWS service provider implementations.
    
    Location: localstack.services.plugins
    """
    service: str  # AWS service name this provider implements

class ServicePlugin:
    """
    Plugin wrapper for service providers.
    
    Location: localstack.services.plugins
    """
    service: str     # Service name
    api: str         # AWS API name
    
    def create_service(self) -> Service:
        """
        Create service instance from this plugin.
        
        Returns:
            Configured Service instance
        """

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

class Skeleton:
    """
    ASF skeleton for AWS service API scaffolding.
    
    Location: localstack.aws.skeleton
    """
    
    def __init__(self, service: 'ServiceModel', delegate):
        """
        Initialize skeleton with service model and implementation delegate.
        
        Args:
            service: AWS service model specification
            delegate: Implementation object for service operations
        """

def load_service(service_name: str) -> 'ServiceModel':
    """
    Load AWS service specification model.
    
    Args:
        service_name: AWS service name (e.g., 's3', 'dynamodb')
        
    Returns:
        ServiceModel with API specifications
        
    Location: localstack.aws.spec
    """

class InternalClientFactory:
    """
    Factory for creating AWS clients connected to LocalStack endpoint.
    
    Location: localstack.aws.connect
    """
    
    def get_client(
        self,
        service_name: str,
        region_name: str = None,
        aws_access_key_id: str = None,
        aws_secret_access_key: str = None,
        aws_session_token: str = None,
        endpoint_url: str = None,
        config = None
    ) -> object:
        """
        Create AWS service client with LocalStack configuration.
        
        Args:
            service_name: AWS service name (e.g., 's3', 'lambda')
            region_name: AWS region (defaults to us-east-1)
            endpoint_url: LocalStack endpoint URL
            **kwargs: Additional boto3 client parameters
            
        Returns:
            Configured boto3 client instance
        """

# Factory instances
connect_to: InternalClientFactory  # For internal LocalStack connections
connect_externally_to: "ExternalClientFactory"  # For external connections

# Core service identifiers
CORE_SERVICES: list[str] = [
    "s3",              # Simple Storage Service
    "lambda",          # Function compute  
    "dynamodb",        # NoSQL database
    "sqs",             # Simple Queue Service
    "sns",             # Simple Notification Service
    "iam",             # Identity and Access Management
    "sts",             # Security Token Service  
    "cloudformation",  # Infrastructure as Code
    "cloudwatch",      # Monitoring and logging
    "apigateway",      # REST/HTTP APIs
]

from localstack.aws.connect import connect_to
import boto3

# Using LocalStack's client factory
s3 = connect_to().s3
dynamodb = connect_to().dynamodb

# Using boto3 directly with LocalStack endpoint
sqs = boto3.client(
    'sqs',
    endpoint_url='http://localhost:4566',
    aws_access_key_id='test',
    aws_secret_access_key='test', 
    region_name='us-east-1'
)

# Create resources
s3.create_bucket(Bucket='my-bucket')
dynamodb.create_table(
    TableName='my-table',
    KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
    AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
    BillingMode='PAY_PER_REQUEST'
)
queue_url = sqs.create_queue(QueueName='my-queue')['QueueUrl']

from localstack.services.plugins import ServiceManager

# Get service manager instance
manager = ServiceManager()

# Check individual service
if manager.is_running("s3"):
    print("S3 service is running")

# Get all service states  
states = manager.get_states()
for service_name, state in states.items():
    print(f"{service_name}: {state.value}")

# Ensure required services are available
try:
    s3_service = manager.require("s3")
    lambda_service = manager.require("lambda")
    print("Required services are available")
except Exception as e:
    print(f"Service requirement not met: {e}")

from localstack.services.plugins import aws_provider, ServiceProvider

@aws_provider(api="myservice")
class MyServiceProvider(ServiceProvider):
    """Custom AWS service provider implementation"""
    
    service = "myservice"
    
    def create_bucket(self, context, request):
        """Example service operation implementation"""
        bucket_name = request["Bucket"]
        # Custom implementation logic
        return {"Location": f"/{bucket_name}"}
    
    def list_buckets(self, context, request):
        """List all buckets"""
        # Custom implementation logic
        return {
            "Buckets": [],
            "Owner": {"ID": "localstack", "DisplayName": "LocalStack"}
        }

# Environment variables for service control
import os

# Enable specific services only
os.environ["SERVICES"] = "s3,lambda,dynamodb"

# Eager loading for faster startup
os.environ["EAGER_SERVICE_LOADING"] = "1"

# Strict service loading validation  
os.environ["STRICT_SERVICE_LOADING"] = "1"

# Service-specific configuration
os.environ["S3_SKIP_SIGNATURE_VALIDATION"] = "1"
os.environ["LAMBDA_EXECUTOR"] = "docker"
os.environ["DYNAMODB_ERROR_PROBABILITY"] = "0.0"

# Health check endpoint
GET /_localstack/health

# Response format:
{
  "services": {
    "s3": "running",
    "lambda": "running",
    "dynamodb": "available", 
    "sqs": "starting"
  }
}

# Control operations
POST /_localstack/health
{
  "action": "restart"  # Control action
}

# LocalStack-specific headers (from localstack.constants)
HEADER_LOCALSTACK_EDGE_URL: str = "x-localstack-edge"           # Proxy forwarding
HEADER_LOCALSTACK_REQUEST_URL: str = "x-localstack-request-url"  # Original URL
HEADER_LOCALSTACK_TARGET: str = "x-localstack-target"        # Target service
HEADER_LOCALSTACK_IDENTIFIER: str = "x-localstack"                      # LocalStack identifier
HEADER_LOCALSTACK_AUTHORIZATION: str = "x-localstack-authorization"  # Custom auth

# AWS-compatible headers
HEADER_AMZN_ERROR_TYPE: str = "X-Amzn-Errortype"             # AWS error type

import requests

# Direct service targeting
response = requests.get(
    'http://localhost:4566/',
    headers={
        'x-localstack-target': 's3',
        'Authorization': 'AWS4-HMAC-SHA256 ...'
    }
)

# Health check
health = requests.get('http://localhost:4566/_localstack/health')
print(health.json())