tessl/pypi-aws-xray-sdk

The AWS X-Ray SDK for Python enables Python developers to record and emit information from within their applications to the AWS X-Ray service for distributed tracing.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

HTTP Utilities

Name: tessl/pypi-aws-xray-sdk
Author: tessl

HTTP-specific metadata collection, request tracing utilities, and trace header propagation for distributed web applications. These utilities help with HTTP request/response tracing and trace context propagation.

Capabilities

HTTP Metadata Collection

Attach HTTP-specific metadata to segments and subsegments for web request tracing.

# Segment/Subsegment methods for HTTP metadata
def put_http_meta(self, key: str, value: Any) -> None:
    """
    Add HTTP-specific metadata to the trace entity.
    
    Args:
        key (str): HTTP metadata key ('url', 'method', 'user_agent', 'client_ip', etc.)
        value (Any): HTTP metadata value
        
    Common Keys:
        - 'url': Request URL
        - 'method': HTTP method (GET, POST, etc.)  
        - 'user_agent': User-Agent header
        - 'client_ip': Client IP address
        - 'x_forwarded_for': X-Forwarded-For header
        - 'request': Request details
        - 'response': Response details
    """

def apply_status_code(self, status_code: int) -> None:
    """
    Apply HTTP status code effects to the trace entity.
    
    Args:
        status_code (int): HTTP status code
        
    Effects:
        - 4xx codes: Sets error flag
        - 5xx codes: Sets fault flag  
        - 429: Sets throttle flag
    """

def set_user(self, user: str) -> None:
    """
    Set user identifier for the segment (indexable in X-Ray console).
    
    Args:
        user (str): User identifier
        
    Notes:
        - Only available on segments, not subsegments
        - Creates searchable annotation in X-Ray console
        - Useful for user-specific trace filtering
    """

Trace Header Management

Handle X-Ray trace headers for distributed tracing across HTTP requests.

class TraceHeader:
    """Represents X-Ray trace header for HTTP request propagation."""
    
    def __init__(self, root: str = None, parent: str = None, sampled: int = None, data: str = None):
        """
        Create a trace header object.
        
        Args:
            root (str): Root trace ID
            parent (str): Parent segment/subsegment ID  
            sampled (int): Sampling decision (0=not sampled, 1=sampled, None=unknown)
            data (str): Additional trace data
        """
    
    @classmethod
    def from_header_str(cls, header: str) -> 'TraceHeader':
        """
        Parse X-Ray trace header string into TraceHeader object.
        
        Args:
            header (str): X-Ray trace header string
            
        Returns:
            TraceHeader: Parsed trace header object
            
        Header Format:
            Root=1-5e8b2e3a-3f1b2c4d5e6f7a8b9c0d1e2f;Parent=1234567890abcdef;Sampled=1
        """
    
    def to_header_str(self) -> str:
        """
        Convert trace header to X-Ray header string format.
        
        Returns:
            str: X-Ray trace header string
        """
    
    # Properties
    @property
    def data(self) -> str:
        """Additional trace data fields."""

Usage Examples

Web Framework Integration

Manual HTTP metadata collection in web applications:

from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core.models.trace_header import TraceHeader

def handle_request(request):
    # Parse incoming trace header
    trace_header_str = request.headers.get('X-Amzn-Trace-Id')
    trace_header = None
    if trace_header_str:
        trace_header = TraceHeader.from_header_str(trace_header_str)
    
    # Create segment with trace context
    with xray_recorder.in_segment('web-request') as segment:
        # Add HTTP request metadata
        segment.put_http_meta('url', request.url)
        segment.put_http_meta('method', request.method)
        segment.put_http_meta('user_agent', request.headers.get('User-Agent'))
        segment.put_http_meta('client_ip', request.remote_addr)
        
        # Set user if authenticated
        if request.user:
            segment.set_user(request.user.id)
        
        # Process request
        response = process_request(request)
        
        # Add HTTP response metadata
        segment.put_http_meta('response', {
            'status': response.status_code
        })
        segment.apply_status_code(response.status_code)
        
        return response

Outgoing HTTP Requests

Add trace headers to outgoing HTTP requests:

import requests
from aws_xray_sdk.core import xray_recorder

def make_api_call(url):
    with xray_recorder.in_subsegment('api-call') as subsegment:
        # Get current trace context
        trace_entity = xray_recorder.get_trace_entity()
        
        # Create trace header for outgoing request
        trace_header = TraceHeader(
            root=trace_entity.trace_id,
            parent=trace_entity.id,
            sampled=1 if trace_entity.sampled else 0
        )
        
        # Add trace header to outgoing request
        headers = {
            'X-Amzn-Trace-Id': trace_header.to_header_str()
        }
        
        # Make request with tracing
        subsegment.put_http_meta('url', url)
        subsegment.put_http_meta('method', 'GET')
        
        response = requests.get(url, headers=headers)
        
        subsegment.put_http_meta('response', {
            'status': response.status_code
        })
        subsegment.apply_status_code(response.status_code)
        
        return response

Error Handling with HTTP Status

Automatic error flagging based on HTTP status codes:

from aws_xray_sdk.core import xray_recorder

def api_endpoint():
    with xray_recorder.in_segment('api-endpoint') as segment:
        try:
            result = process_api_request()
            segment.apply_status_code(200)
            return result, 200
        except ValidationError:
            segment.apply_status_code(400)  # Sets error flag
            return {'error': 'Invalid input'}, 400
        except Exception as e:
            segment.apply_status_code(500)  # Sets fault flag
            segment.add_exception(e)
            return {'error': 'Internal error'}, 500