CtrlK
BlogDocsLog inGet started
Tessl Logo

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

web-frameworks.mddocs/

Web Framework Integration

Middleware and integration components for popular Python web frameworks including Django, Flask, Bottle, and aiohttp. Provides automatic request tracing, HTTP metadata collection, and seamless integration with framework-specific features.

Capabilities

Django Integration

Complete Django integration with middleware, settings configuration, and automatic request tracing.

Django Middleware

# Settings configuration
INSTALLED_APPS = [
    'aws_xray_sdk.ext.django',
    # ... other apps
]

MIDDLEWARE = [
    'aws_xray_sdk.ext.django.middleware.XRayMiddleware',
    # ... other middleware
]

# X-Ray configuration in settings
XRAY_RECORDER = {
    'AWS_XRAY_TRACING_NAME': str,           # Required: segment name for requests
    'PLUGINS': tuple,                        # Optional: plugin names
    'DAEMON_ADDRESS': str,                   # Optional: daemon address
    'CONTEXT_MISSING': str,                  # Optional: context missing behavior
    'SAMPLING': bool,                        # Optional: sampling enabled
    'SAMPLING_RULES': str,                   # Optional: sampling rules file path
    'PATCH_MODULES': list,                   # Optional: modules to patch
    'IGNORE_MODULE_PATTERNS': list,          # Optional: module patterns to ignore
    'AUTO_PATCH_PARENT_SEGMENT_NAME': str,   # Optional: parent segment for auto-patching
    'STREAM_SQL': bool,                      # Optional: SQL streaming enabled
    'AUTO_INSTRUMENT': bool,                 # Optional: auto-instrument Django ORM
    'URLS_AS_ANNOTATION': str,               # Optional: 'LAMBDA', 'ALL', or 'NONE'
}

Django Database Integration

# Auto-instrument Django ORM
XRAY_RECORDER = {
    'AUTO_INSTRUMENT': True,  # Automatically patch Django database operations
    'STREAM_SQL': True,       # Enable SQL query streaming
}

# Manual database patching
from aws_xray_sdk.ext.django.db import patch_db
patch_db()

Flask Integration

Flask middleware for automatic request tracing and X-Ray integration.

class XRayMiddleware:
    """Flask middleware for X-Ray request tracing."""
    
    def __init__(self, app: Flask, recorder: AWSXRayRecorder) -> None:
        """
        Initialize Flask X-Ray middleware.

        Args:
            app (Flask): Flask application instance
            recorder (AWSXRayRecorder): X-Ray recorder instance
        """

Bottle Integration

Bottle plugin for X-Ray request tracing integration.

class XRayMiddleware:
    """Bottle plugin for X-Ray request tracing."""
    
    def __init__(self, recorder: AWSXRayRecorder) -> None:
        """
        Initialize Bottle X-Ray plugin.

        Args:
            recorder (AWSXRayRecorder): X-Ray recorder instance
        """

aiohttp Integration

Asynchronous web framework integration with middleware and client tracing.

aiohttp Middleware

def middleware(request: web.Request, handler: Callable) -> web.Response:
    """
    aiohttp middleware function for X-Ray request tracing.

    Args:
        request (web.Request): aiohttp request object
        handler (Callable): Request handler function

    Returns:
        web.Response: Response object with X-Ray tracing
    """

aiohttp Client Tracing

def aws_xray_trace_config() -> aiohttp.TraceConfig:
    """
    Create aiohttp TraceConfig for client request tracing.

    Returns:
        aiohttp.TraceConfig: Configured trace config for X-Ray integration
    """

Framework-Specific Usage

Django Setup

Basic Django Configuration

# settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'aws_xray_sdk.ext.django',  # Add X-Ray Django extension
    'myapp',
]

MIDDLEWARE = [
    'aws_xray_sdk.ext.django.middleware.XRayMiddleware',  # Add as first middleware
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    # ... other middleware
]

# X-Ray configuration
XRAY_RECORDER = {
    'AWS_XRAY_TRACING_NAME': 'My Django App',  # Required
    'PLUGINS': ('EC2Plugin', 'ECSPlugin'),
    'DAEMON_ADDRESS': '127.0.0.1:2000',
    'CONTEXT_MISSING': 'LOG_ERROR',
    'SAMPLING': False,  # Disable sampling for development
}

Advanced Django Configuration

# settings.py
XRAY_RECORDER = {
    'AWS_XRAY_TRACING_NAME': 'Django Production App',
    'PLUGINS': ('EC2Plugin', 'ECSPlugin', 'ElasticBeanstalkPlugin'),
    'DAEMON_ADDRESS': '127.0.0.1:2000',
    'CONTEXT_MISSING': 'LOG_ERROR',
    'SAMPLING': True,
    'SAMPLING_RULES': '/path/to/sampling-rules.json',
    
    # Automatic patching
    'PATCH_MODULES': [
        'boto3',
        'requests',
        'sqlalchemy_core',
        'pymongo',
    ],
    'IGNORE_MODULE_PATTERNS': [
        'myapp.test.*',
        '.*mock.*',
    ],
    'AUTO_PATCH_PARENT_SEGMENT_NAME': 'Django App Initialization',
    
    # Database integration
    'AUTO_INSTRUMENT': True,  # Auto-patch Django ORM
    'STREAM_SQL': True,       # Stream SQL queries
    
    # Lambda-specific settings
    'URLS_AS_ANNOTATION': 'LAMBDA',  # Add URL as annotation for Lambda
}

Django Views with X-Ray

# views.py
from django.http import JsonResponse
from aws_xray_sdk.core import xray_recorder

def user_profile(request, user_id):
    # Request is automatically traced by middleware
    
    with xray_recorder.in_subsegment('validate-user') as subsegment:
        subsegment.put_annotation('user_id', user_id)
        # Validate user logic
    
    with xray_recorder.in_subsegment('fetch-profile') as subsegment:
        subsegment.put_annotation('table', 'user_profiles')
        # Database query - automatically traced if AUTO_INSTRUMENT=True
        profile = UserProfile.objects.get(user_id=user_id)
    
    return JsonResponse({'profile': profile.to_dict()})

# Class-based views
from django.views.generic import View

class UserAPIView(View):
    @xray_recorder.capture('process-user-request')
    def get(self, request, user_id):
        # Method is automatically traced
        return JsonResponse({'user_id': user_id})

Flask Setup

Basic Flask Integration

from flask import Flask
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware

app = Flask(__name__)

# Configure X-Ray recorder
xray_recorder.configure(
    service='My Flask App',
    dynamic_naming='*.example.com*',
    daemon_address='127.0.0.1:2000',
    context_missing='LOG_ERROR'
)

# Add X-Ray middleware
XRayMiddleware(app, xray_recorder)

@app.route('/users/<user_id>')
def get_user(user_id):
    # Request is automatically traced
    
    with xray_recorder.in_subsegment('database-query') as subsegment:
        subsegment.put_annotation('user_id', user_id)
        # Database operation
        user = fetch_user_from_db(user_id)
    
    return {'user': user}

if __name__ == '__main__':
    app.run()

Flask with Database Integration

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware
from aws_xray_sdk.ext.flask_sqlalchemy.query import XRayFlaskSqlAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'

# Configure X-Ray
xray_recorder.configure(service='Flask App with DB')
XRayMiddleware(app, xray_recorder)

# Use X-Ray enabled SQLAlchemy
db = XRayFlaskSqlAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)

@app.route('/users/<int:user_id>')
def get_user(user_id):
    # Database queries are automatically traced
    user = User.query.get_or_404(user_id)
    return {'id': user.id, 'username': user.username}

Bottle Setup

from bottle import Bottle, run, route
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.ext.bottle.middleware import XRayMiddleware

app = Bottle()

# Configure X-Ray recorder
xray_recorder.configure(
    service='My Bottle App',
    dynamic_naming='*.example.com*'
)

# Install X-Ray plugin
app.install(XRayMiddleware(xray_recorder))

@app.route('/users/<user_id>')
def get_user(user_id):
    # Request is automatically traced
    
    with xray_recorder.in_subsegment('process-user') as subsegment:
        subsegment.put_annotation('user_id', user_id)
        # Process user logic
    
    return {'user_id': user_id}

if __name__ == '__main__':
    run(app, host='localhost', port=8080)

aiohttp Setup

aiohttp Server Integration

from aiohttp import web
from aws_xray_sdk.ext.aiohttp.middleware import middleware
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core.async_context import AsyncContext

# Configure recorder with async context
xray_recorder.configure(
    service='aiohttp App',
    context=AsyncContext()
)

async def hello_handler(request):
    # Request is automatically traced
    
    async with xray_recorder.in_subsegment_async('async-operation') as subsegment:
        subsegment.put_annotation('operation', 'hello')
        # Async operation
        await asyncio.sleep(0.1)
    
    return web.json_response({'message': 'Hello World'})

# Create application with X-Ray middleware
app = web.Application(middlewares=[middleware])
app.router.add_get('/', hello_handler)

if __name__ == '__main__':
    web.run_app(app)

aiohttp Client Tracing

import aiohttp
from aws_xray_sdk.ext.aiohttp.client import aws_xray_trace_config
from aws_xray_sdk.core import xray_recorder

async def make_http_requests():
    # Create trace config for client requests
    trace_config = aws_xray_trace_config()
    
    async with aiohttp.ClientSession(trace_configs=[trace_config]) as session:
        # Client requests are automatically traced
        async with session.get('https://api.example.com/users') as response:
            data = await response.json()
            return data

# Use within a traced context
async def main():
    async with xray_recorder.in_segment_async('http-client-demo') as segment:
        data = await make_http_requests()
        segment.put_metadata('response_data', data)

Combined Framework and Database Integration

from flask import Flask
from aws_xray_sdk.core import xray_recorder, patch_all
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware
import requests
import boto3

app = Flask(__name__)

# Configure X-Ray
xray_recorder.configure(
    service='Integrated Flask App',
    plugins=('EC2Plugin',),
    daemon_address='127.0.0.1:2000'
)

# Patch all supported libraries
patch_all()

# Add Flask middleware
XRayMiddleware(app, xray_recorder)

@app.route('/user-data/<user_id>')
def get_user_data(user_id):
    # Request automatically traced by middleware
    
    with xray_recorder.in_subsegment('fetch-user-profile') as subsegment:
        subsegment.put_annotation('user_id', user_id)
        # HTTP request - automatically traced via patching
        response = requests.get(f'https://api.users.com/profile/{user_id}')
        profile = response.json()
    
    with xray_recorder.in_subsegment('fetch-user-preferences') as subsegment:
        # AWS call - automatically traced via patching
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table('user_preferences')
        preferences = table.get_item(Key={'user_id': user_id})
    
    return {
        'profile': profile,
        'preferences': preferences.get('Item', {})
    }

Serverless Integration

Lambda with Web Frameworks

X-Ray works seamlessly with web frameworks deployed on AWS Lambda, automatically creating subsegments within Lambda's managed segments.

Django on Lambda

# serverless Django setup
import os
os.environ['AWS_XRAY_TRACING_NAME'] = 'Django Lambda'

# Django settings remain the same
# Middleware will create subsegments instead of segments in Lambda

Flask on Lambda

from flask import Flask
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware

app = Flask(__name__)

# Configure for Lambda environment
xray_recorder.configure(
    service='Flask Lambda Function',
    context_missing='LOG_ERROR'
)

# Middleware creates subsegments in Lambda
XRayMiddleware(app, xray_recorder)

def lambda_handler(event, context):
    # Lambda integration handles this automatically
    return app(event, context)

Best Practices

Middleware Ordering

# Django - X-Ray middleware should be first
MIDDLEWARE = [
    'aws_xray_sdk.ext.django.middleware.XRayMiddleware',  # First
    'django.middleware.security.SecurityMiddleware',
    # ... other middleware
]

Error Handling

# Automatic error capture in web frameworks
@app.errorhandler(500)
def handle_error(error):
    # Error automatically captured by X-Ray middleware
    # Additional context can be added
    xray_recorder.put_annotation('error_handler', 'custom_500')
    return 'Internal Server Error', 500

Performance Considerations

# Use sampling for high-traffic applications
xray_recorder.configure(
    sampling=True,  # Enable sampling
    sampling_rules='/path/to/sampling-rules.json'
)

# Conditional expensive operations
if xray_recorder.is_sampled():
    # Only add expensive metadata for sampled requests
    xray_recorder.put_metadata('detailed_request_info', expensive_operation())

Install with Tessl CLI

npx tessl i tessl/pypi-aws-xray-sdk

docs

annotations-metadata.md

aws-services.md

configuration-plugins.md

core-recording.md

database-integration.md

http-utilities.md

index.md

library-patching.md

sampling.md

web-frameworks.md

tile.json