Annotations and Metadata

def put_annotation(key: str, value: str) -> None:
    """
    Add an annotation to the current active trace entity.

    Args:
        key (str): Annotation key (must be a string)
        value (str): Annotation value (must be a string)

    Notes:
        - Annotations are indexed and searchable in X-Ray console
        - Maximum 50 annotations per segment/subsegment
        - Key and value must be strings
        - Used for filtering traces in X-Ray console queries
    """

def put_metadata(key: str, value: Any, namespace: str = 'default') -> None:
    """
    Add metadata to the current active trace entity.

    Args:
        key (str): Metadata key
        value (Any): Metadata value (can be any JSON-serializable type)
        namespace (str): Optional namespace for organizing metadata (default: 'default')

    Notes:
        - Metadata is not indexed but provides rich debugging context
        - Can store complex objects, lists, dictionaries
        - Organized by namespaces for better organization
        - No limit on metadata size (within segment size limits)
    """

def is_sampled() -> bool:
    """
    Check if the current trace entity is being sampled.

    Returns:
        bool: True if the current entity is sampled, False otherwise

    Notes:
        - Use to conditionally execute expensive annotation/metadata generation
        - Unsampled traces still record structure but skip detailed data
    """

class Segment:
    def put_annotation(self, key: str, value: str) -> None:
        """Add annotation directly to this segment."""
    
    def put_metadata(self, key: str, value: Any, namespace: str = 'default') -> None:
        """Add metadata directly to this segment."""

class Subsegment:
    def put_annotation(self, key: str, value: str) -> None:
        """Add annotation directly to this subsegment."""
    
    def put_metadata(self, key: str, value: Any, namespace: str = 'default') -> None:
        """Add metadata directly to this subsegment."""

def put_http_meta(key: str, value: Any) -> None:
    """
    Add HTTP-specific metadata to the current trace entity.

    Args:
        key (str): HTTP metadata key ('url', 'method', 'status', 'user_agent', etc.)
        value (Any): HTTP metadata value

    Common Keys:
        - 'url': Request URL
        - 'method': HTTP method (GET, POST, etc.)
        - 'status': HTTP status code
        - 'user_agent': Client user agent
        - 'client_ip': Client IP address
        - 'content_length': Response content length
    """

def add_throttle_flag() -> None:
    """Mark the current entity as throttled."""

def add_fault_flag() -> None:
    """Mark the current entity as having a fault (5xx errors)."""

def add_error_flag() -> None:
    """Mark the current entity as having an error (4xx errors)."""

def apply_status_code(status_code: int) -> None:
    """
    Apply HTTP status code effects to the current entity.

    Args:
        status_code (int): HTTP status code

    Effects:
        - 4xx codes: Adds error flag
        - 5xx codes: Adds fault flag
        - 429 code: Adds throttle flag
    """

def add_exception(exception: Exception, stack: list = None, remote: bool = False) -> None:
    """
    Add exception information to the current trace entity.

    Args:
        exception (Exception): Exception object to record
        stack (list): Optional stack trace (auto-generated if not provided)
        remote (bool): Whether the exception originated from a remote service

    Notes:
        - Automatically called when exceptions occur within traced code
        - Can be called manually for custom exception handling
        - Records exception type, message, and stack trace
    """

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('user-request') as segment:
    # Add searchable annotations
    xray_recorder.put_annotation('user_id', '12345')
    xray_recorder.put_annotation('user_type', 'premium')
    xray_recorder.put_annotation('region', 'us-east-1')
    
    # These annotations can be used for filtering in X-Ray console:
    # annotation.user_type = "premium"
    # annotation.region = "us-east-1"

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('api-request') as segment:
    # Add structured metadata
    xray_recorder.put_metadata('request_info', {
        'method': 'POST',
        'path': '/api/users',
        'query_params': {'limit': 50, 'offset': 0},
        'headers': {'user-agent': 'MyApp/1.0'},
        'payload_size': 1024
    })
    
    # Add metadata in different namespaces
    xray_recorder.put_metadata('database_config', {
        'host': 'db.example.com',
        'database': 'users',
        'connection_pool_size': 10
    }, namespace='database')
    
    xray_recorder.put_metadata('feature_flags', {
        'new_ui': True,
        'advanced_search': False,
        'beta_features': True
    }, namespace='configuration')

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('data-processing') as segment:
    # Only perform expensive operations if trace is sampled
    if xray_recorder.is_sampled():
        # Generate expensive debugging information
        performance_metrics = analyze_performance_deep()
        system_state = capture_system_state()
        
        xray_recorder.put_metadata('performance', performance_metrics)
        xray_recorder.put_metadata('system_state', system_state)
        
        # Generate detailed annotations
        xray_recorder.put_annotation('cpu_usage', str(performance_metrics['cpu']))
        xray_recorder.put_annotation('memory_usage', str(performance_metrics['memory']))
    else:
        # Minimal annotation for unsampled traces
        xray_recorder.put_annotation('operation', 'data_processing')

from aws_xray_sdk.core import xray_recorder
import requests

with xray_recorder.in_segment('web-request') as segment:
    # Add HTTP-specific metadata
    segment.put_http_meta('method', 'GET')
    segment.put_http_meta('url', 'https://api.example.com/users')
    segment.put_http_meta('user_agent', 'MyApp/1.0')
    
    try:
        response = requests.get('https://api.example.com/users')
        
        # Add response information
        segment.put_http_meta('status', response.status_code)
        segment.put_http_meta('content_length', len(response.content))
        
        # Apply status code effects
        segment.apply_status_code(response.status_code)
        
        # Add business logic annotations
        xray_recorder.put_annotation('api_version', response.headers.get('API-Version', 'unknown'))
        xray_recorder.put_annotation('response_size', str(len(response.content)))
        
    except requests.RequestException as e:
        # Exception is automatically captured
        xray_recorder.put_annotation('error_type', 'network_error')
        raise

from aws_xray_sdk.core import xray_recorder
import sqlite3

with xray_recorder.in_segment('database-operation') as segment:
    with xray_recorder.in_subsegment('user-query') as subsegment:
        # Add database-specific annotations
        subsegment.put_annotation('table', 'users')
        subsegment.put_annotation('operation', 'SELECT')
        subsegment.put_annotation('index_used', 'user_id_idx')
        
        # Add detailed metadata
        subsegment.put_metadata('query_info', {
            'sql': 'SELECT * FROM users WHERE user_id = ?',
            'parameters': ['12345'],
            'estimated_rows': 1,
            'execution_plan': 'index_scan'
        }, namespace='database')
        
        # Execute query
        conn = sqlite3.connect('app.db')
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM users WHERE user_id = ?', ('12345',))
        result = cursor.fetchone()
        
        # Add result metadata
        subsegment.put_metadata('result_info', {
            'rows_returned': 1 if result else 0,
            'execution_time_ms': 15.2,
            'cache_hit': False
        }, namespace='database')

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('error-prone-operation') as segment:
    try:
        # Risky operation
        result = perform_risky_operation()
        
        # Success annotations
        xray_recorder.put_annotation('status', 'success')
        
    except ValueError as e:
        # Client error (4xx equivalent)
        segment.add_error_flag()
        xray_recorder.put_annotation('error_type', 'validation_error')
        xray_recorder.put_metadata('error_details', {
            'error_message': str(e),
            'error_code': 'INVALID_INPUT'
        })
        raise
        
    except ConnectionError as e:
        # Server fault (5xx equivalent)
        segment.add_fault_flag()
        xray_recorder.put_annotation('error_type', 'connection_error')
        raise
        
    except Exception as e:
        # Unknown error
        segment.add_fault_flag()
        xray_recorder.put_annotation('error_type', 'unknown_error')
        # Exception details are automatically captured
        raise

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('multi-step-process') as segment:
    # Segment-level metadata
    segment.put_annotation('process_type', 'batch_job')
    segment.put_metadata('job_config', {
        'batch_size': 1000,
        'timeout': 300,
        'retry_count': 3
    })
    
    for i in range(3):
        with xray_recorder.in_subsegment(f'step-{i+1}') as subsegment:
            # Subsegment-specific metadata
            subsegment.put_annotation('step_number', str(i+1))
            subsegment.put_metadata('step_config', {
                'input_size': 100 * (i+1),
                'processing_mode': 'parallel' if i > 0 else 'sequential'
            })
            
            # Process step
            process_step(i+1)

from aws_xray_sdk.core import xray_recorder

with xray_recorder.in_segment('optimized-operation') as segment:
    # Always add essential annotations
    xray_recorder.put_annotation('operation_id', operation_id)
    
    # Conditionally add expensive metadata
    if xray_recorder.is_sampled():
        # Expensive operations only for sampled traces
        detailed_metrics = generate_detailed_metrics()
        xray_recorder.put_metadata('detailed_metrics', detailed_metrics)
    
    # Lightweight annotations for all traces
    xray_recorder.put_annotation('status', 'processed')