Library Patching

def patch_all(double_patch: bool = False) -> None:
    """
    Patch all supported libraries for X-Ray tracing.

    Args:
        double_patch (bool): Enable patching of indirect dependencies.
                           When True, patches libraries that other libraries depend on.

    Supported Libraries:
        - aiobotocore: Async AWS SDK
        - botocore: AWS SDK (boto3)
        - pynamodb: DynamoDB ORM
        - requests: HTTP library
        - sqlite3: SQLite database
        - mysql: MySQL database
        - httplib: Standard HTTP library
        - pymongo: MongoDB driver
        - pymysql: Pure Python MySQL driver
        - psycopg2: PostgreSQL adapter
        - pg8000: Pure Python PostgreSQL driver
        - sqlalchemy_core: SQLAlchemy Core
        - httpx: Modern async HTTP client
    """

def patch(
    modules_to_patch: tuple,
    raise_errors: bool = True,
    ignore_module_patterns: list = None
) -> None:
    """
    Patch specific modules for X-Ray tracing.

    Args:
        modules_to_patch (tuple): Tuple of module names to patch
        raise_errors (bool): Whether to raise errors if patching fails
        ignore_module_patterns (list): List of regex patterns for modules to ignore

    Raises:
        ImportError: If module cannot be imported and raise_errors is True
        Exception: If patching fails and raise_errors is True
    """

from aws_xray_sdk.core import patch
import boto3

# Patch botocore for AWS service tracing
patch(['botocore'])

# All AWS calls are now automatically traced
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('users')
response = table.get_item(Key={'id': '123'})  # Automatically traced

from aws_xray_sdk.core import patch
import aiobotocore.session

# Patch aiobotocore for async AWS service tracing
patch(['aiobotocore'])

# Async AWS calls are automatically traced
session = aiobotocore.session.get_session()
async with session.create_client('s3') as client:
    response = await client.list_buckets()  # Automatically traced

from aws_xray_sdk.core import patch
import requests

# Patch requests for HTTP tracing
patch(['requests'])

# HTTP requests are automatically traced
response = requests.get('https://api.example.com/data')  # Automatically traced
response = requests.post('https://api.example.com/users', json={'name': 'John'})  # Automatically traced

from aws_xray_sdk.core import patch
import http.client

# Patch httplib for HTTP tracing
patch(['httplib'])

# Standard library HTTP calls are traced
conn = http.client.HTTPSConnection('api.example.com')
conn.request('GET', '/data')
response = conn.getresponse()  # Automatically traced

from aws_xray_sdk.ext.httplib import add_ignored, reset_ignored

# Ignore requests to specific hostnames
add_ignored(hostname='test.myapp.com')

# Ignore requests with glob patterns
add_ignored(hostname='*.internal.com')

# Ignore specific URL paths
add_ignored(urls=['/health', '/metrics'])

# Ignore requests from specific HTTP client classes
add_ignored(subclass='botocore.awsrequest.AWSHTTPConnection')

# Combine multiple filters
add_ignored(hostname='api.example.com', urls=['/internal/*'])

# Reset all ignore patterns to defaults
reset_ignored()

from aws_xray_sdk.core import patch
import httpx

# Patch httpx for HTTP tracing
patch(['httpx'])

# Both sync and async httpx calls are traced
response = httpx.get('https://api.example.com/data')  # Automatically traced

async with httpx.AsyncClient() as client:
    response = await client.get('https://api.example.com/data')  # Automatically traced

from aws_xray_sdk.core import patch
import sqlite3

# Patch sqlite3 for database tracing
patch(['sqlite3'])

# SQLite operations are automatically traced
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')  # Automatically traced

from aws_xray_sdk.core import patch

# Patch MySQL drivers
patch(['mysql'])  # MySQLdb
patch(['pymysql'])  # PyMySQL

import MySQLdb
# or
import pymysql

# Database operations are automatically traced

from aws_xray_sdk.core import patch

# Patch PostgreSQL drivers
patch(['psycopg2'])  # psycopg2
patch(['pg8000'])    # pg8000

import psycopg2
# or
import pg8000

# Database operations are automatically traced

from aws_xray_sdk.core import patch
import pymongo

# Patch pymongo for MongoDB tracing
patch(['pymongo'])

# MongoDB operations are automatically traced
client = pymongo.MongoClient()
db = client.mydb
collection = db.users
result = collection.find_one({'name': 'John'})  # Automatically traced

from aws_xray_sdk.core import patch
from sqlalchemy import create_engine

# Patch SQLAlchemy Core
patch(['sqlalchemy_core'])

# SQLAlchemy operations are automatically traced
engine = create_engine('postgresql://user:pass@localhost/db')
result = engine.execute('SELECT * FROM users')  # Automatically traced

from aws_xray_sdk.core import patch
from pynamodb.models import Model

# Patch PynamoDB
patch(['pynamodb'])

# PynamoDB operations are automatically traced
class User(Model):
    class Meta:
        table_name = 'users'
    
    id = UnicodeAttribute(hash_key=True)
    name = UnicodeAttribute()

user = User.get('123')  # Automatically traced

from aws_xray_sdk.core import patch_all, patch, xray_recorder

# Configure recorder first
xray_recorder.configure(
    sampling=False,
    context_missing='LOG_ERROR'
)

# Option 1: Patch all supported libraries
patch_all()

# Option 2: Patch specific libraries
patch(['requests', 'botocore', 'sqlite3'])

# Your application code - all patched libraries are now traced
import requests
import boto3

with xray_recorder.in_segment('my-application'):
    # These calls are automatically traced
    response = requests.get('https://api.example.com')
    s3 = boto3.client('s3')
    buckets = s3.list_buckets()

from aws_xray_sdk.core import patch

# Patch with error handling
try:
    patch(['requests', 'botocore', 'nonexistent_library'], raise_errors=False)
    print("Patching completed - some modules may have been skipped")
except Exception as e:
    print(f"Patching failed: {e}")

# Patch with module filtering
patch(
    ['requests', 'botocore'],
    ignore_module_patterns=[r'.*test.*', r'.*mock.*']
)

# For Django applications
from aws_xray_sdk.core import patch
patch(['requests', 'botocore'])

# Django middleware handles segment creation
# Add to settings.py:
# MIDDLEWARE = ['aws_xray_sdk.ext.django.middleware.XRayMiddleware', ...]

# For Flask applications  
from aws_xray_sdk.core import patch, xray_recorder
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware

patch(['requests', 'botocore'])
app = Flask(__name__)
XRayMiddleware(app, xray_recorder)

from aws_xray_sdk.core import patch

# Ignore test and development modules
patch(
    ['requests', 'botocore'],
    ignore_module_patterns=[
        r'.*test.*',      # Ignore any module with 'test' in the name
        r'.*mock.*',      # Ignore mock modules
        r'dev\..*',       # Ignore dev.* modules
        r'.*\.test_.*'    # Ignore test_ modules
    ]
)

from aws_xray_sdk.core import patch_all

# Enable double patching for indirect dependencies
patch_all(double_patch=True)

from aws_xray_sdk.core import patch

# Patch entire local modules recursively
patch(['myapp.services', 'myapp.utils'])

# This applies @xray_recorder.capture() decorator to all functions
# and class methods in the specified modules and their submodules

from aws_xray_sdk.core import patch_all, xray_recorder

# 1. Configure recorder first
xray_recorder.configure(
    sampling=False,
    context_missing='LOG_ERROR',
    daemon_address='127.0.0.1:2000'
)

# 2. Patch libraries early in application startup
patch_all()

# 3. Import libraries after patching
import requests
import boto3
import sqlite3

# 4. Use with proper segment management
with xray_recorder.in_segment('application-startup'):
    # Initialize your application
    pass

from aws_xray_sdk.core import xray_recorder

# Check if a library is properly patched
if hasattr(requests.get, '__wrapped__'):
    print("Requests is properly patched")

# Test tracing is working
with xray_recorder.in_segment('test-segment'):
    response = requests.get('https://httpbin.org/get')
    print(f"Request traced: {response.status_code}")