Context Management

from raven.context import Context

class Context(local, Mapping, Iterable):
    def __init__(self, client=None):
        """
        Initialize context with empty data storage.
        
        Parameters:
        - client (Client): Associated Sentry client instance
        """

    def activate(self, sticky=False):
        """
        Activate context for current thread.

        Parameters:
        - sticky (bool): Keep context active after deactivation
        """

    def deactivate(self):
        """Deactivate context for current thread."""

    def merge(self, data, activate=True):
        """
        Merge data into context.

        Parameters:
        - data (dict): Context data to merge
        - activate (bool): Activate context after merging

        Returns:
        Context: Self for chaining
        """

    def clear(self, deactivate=None):
        """
        Clear context data.

        Parameters:
        - deactivate (bool): Whether to deactivate context
        """

    def get(self):
        """
        Get current context data.

        Returns:
        dict: Current context data
        """

    def set(self, data):
        """
        Set context data, replacing existing data.

        Parameters:
        - data (dict): Context data to set
        """

    def __enter__(self):
        """Context manager entry - activates context."""

    def __exit__(self, exc_type, exc_value, tb):
        """Context manager exit - deactivates context."""

    def __getitem__(self, key):
        """Get context data by key (dict-like access)."""

    def __iter__(self):
        """Iterate over context keys."""

    def __len__(self):
        """Get number of context items."""

from raven.context import get_active_contexts

def get_active_contexts():
    """
    Get all active contexts for current thread.

    Returns:
    list: List of active Context instances
    """

# From Client class
def user_context(self, data):
    """
    Set user context information.

    Parameters:
    - data (dict): User data including 'id', 'username', 'email', 'ip_address'
    """

def tags_context(self, data):
    """
    Set tags for event categorization and filtering.

    Parameters:
    - data (dict): Tag key-value pairs
    """

def extra_context(self, data):
    """
    Set extra context data for debugging.

    Parameters:
    - data (dict): Additional context information
    """

def http_context(self, data):
    """
    Set HTTP request context.

    Parameters:
    - data (dict): HTTP context including 'url', 'method', 'headers', 'query_string'
    """

from raven import Client
from raven.context import Context

client = Client('https://your-dsn@sentry.io/project-id')

# Set user context
client.user_context({
    'id': 123,
    'username': 'john_doe',
    'email': 'john@example.com',
    'ip_address': '192.168.1.1'
})

# Set tags for categorization
client.tags_context({
    'environment': 'production',
    'version': '1.2.3',
    'feature_flag': 'new_checkout',
    'user_type': 'premium'
})

# Set extra debugging information
client.extra_context({
    'request_id': 'req_abc123',
    'session_id': 'sess_xyz789',
    'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'referrer': 'https://example.com/products'
})

# This exception will include all the context above
try:
    process_checkout()
except Exception:
    client.captureException()

from flask import Flask, request
from raven.contrib.flask import Sentry

app = Flask(__name__)
sentry = Sentry(app)

@app.before_request
def set_request_context():
    # Set context for each request
    sentry.client.user_context({
        'ip_address': request.remote_addr
    })
    
    sentry.client.tags_context({
        'endpoint': request.endpoint,
        'method': request.method
    })
    
    sentry.client.http_context({
        'url': request.url,
        'method': request.method,
        'headers': dict(request.headers),
        'query_string': request.query_string.decode()
    })

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    # Add user-specific context
    sentry.client.user_context({'id': user_id})
    sentry.client.tags_context({'operation': 'get_user'})
    
    try:
        user = User.get(user_id)
        return jsonify(user.to_dict())
    except UserNotFound:
        # This will include all the request context
        sentry.captureException()
        return jsonify({'error': 'User not found'}), 404

from raven.context import Context

def process_user_request(user_id, request_data):
    context = Context()
    
    # Activate context for this operation
    context.activate()
    
    try:
        # Merge user-specific data
        context.merge({
            'user': {'id': user_id},
            'tags': {'operation': 'user_request'},
            'extra': {'request_size': len(str(request_data))}
        })
        
        # Process request - any errors will include context
        result = complex_operation(request_data)
        
        return result
        
    finally:
        # Clean up context
        context.deactivate()

import threading
from raven import Client
from raven.context import Context

client = Client('https://your-dsn@sentry.io/project-id')

def worker_thread(task_id, user_id):
    # Create context for this thread
    context = Context()
    context.activate()
    
    # Set thread-specific context
    context.merge({
        'user': {'id': user_id},
        'tags': {
            'task_id': task_id,
            'thread': threading.current_thread().name
        },
        'extra': {'start_time': time.time()}
    })
    
    try:
        # This work will have the thread context
        result = process_task(task_id)
        client.captureMessage(f'Task {task_id} completed successfully')
        return result
        
    except Exception:
        # Exception will include thread context
        client.captureException()
        raise
    finally:
        context.deactivate()

# Main thread context
client.user_context({'id': 'system'})
client.tags_context({'component': 'task_manager'})

# Spawn worker threads
threads = []
for i in range(5):
    t = threading.Thread(
        target=worker_thread,
        args=(f'task_{i}', f'user_{i}')
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

from raven import Client

class SentryContextMiddleware:
    def __init__(self, client):
        self.client = client
    
    def __call__(self, environ, start_response):
        # Extract request information
        request_method = environ.get('REQUEST_METHOD')
        path_info = environ.get('PATH_INFO')
        query_string = environ.get('QUERY_STRING')
        remote_addr = environ.get('REMOTE_ADDR')
        
        # Set request context
        self.client.http_context({
            'url': f"{path_info}?{query_string}" if query_string else path_info,
            'method': request_method,
            'headers': self._get_headers(environ),
            'query_string': query_string,
        })
        
        self.client.tags_context({
            'path': path_info,
            'method': request_method
        })
        
        self.client.extra_context({
            'remote_addr': remote_addr,
            'server_name': environ.get('SERVER_NAME'),
            'server_port': environ.get('SERVER_PORT')
        })
        
        try:
            return self.app(environ, start_response)
        except Exception:
            self.client.captureException()
            raise
    
    def _get_headers(self, environ):
        headers = {}
        for key, value in environ.items():
            if key.startswith('HTTP_'):
                header_name = key[5:].replace('_', '-').title()
                headers[header_name] = value
        return headers

# Use with WSGI app
app = get_wsgi_app()
client = Client('https://your-dsn@sentry.io/project-id')
app = SentryContextMiddleware(client)(app)

from raven import Client

class ContextualClient:
    def __init__(self, dsn):
        self.client = Client(dsn)
        self._context_stack = []
    
    def push_context(self, **context_data):
        """Push new context data onto stack."""
        self._context_stack.append(context_data)
        self._update_client_context()
    
    def pop_context(self):
        """Pop most recent context data from stack."""
        if self._context_stack:
            self._context_stack.pop()
            self._update_client_context()
    
    def _update_client_context(self):
        """Update client with merged context from stack."""
        merged_context = {}
        for context in self._context_stack:
            for key, value in context.items():
                if key in merged_context and isinstance(value, dict):
                    merged_context[key].update(value)
                else:
                    merged_context[key] = value
        
        # Apply to client
        if 'user' in merged_context:
            self.client.user_context(merged_context['user'])
        if 'tags' in merged_context:
            self.client.tags_context(merged_context['tags'])
        if 'extra' in merged_context:
            self.client.extra_context(merged_context['extra'])
    
    def capture_exception(self, **kwargs):
        return self.client.captureException(**kwargs)
    
    def capture_message(self, message, **kwargs):
        return self.client.captureMessage(message, **kwargs)

# Usage
contextual_client = ContextualClient('https://your-dsn@sentry.io/project-id')

# Set base context
contextual_client.push_context(
    user={'id': 123},
    tags={'component': 'auth'}
)

# Add operation-specific context
contextual_client.push_context(
    tags={'operation': 'login'},
    extra={'login_method': 'password'}
)

try:
    authenticate_user()
except AuthError:
    # Captures with merged context
    contextual_client.capture_exception()

# Remove operation context, keep base context
contextual_client.pop_context()

from functools import wraps
from raven import Client

client = Client('https://your-dsn@sentry.io/project-id')

def with_context(**context_data):
    """Decorator to add context to function execution."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Store original context
            original_user = getattr(client.context, 'user', {})
            original_tags = getattr(client.context, 'tags', {})
            original_extra = getattr(client.context, 'extra', {})
            
            try:
                # Apply decorator context
                if 'user' in context_data:
                    client.user_context(context_data['user'])
                if 'tags' in context_data:
                    client.tags_context(context_data['tags'])
                if 'extra' in context_data:
                    client.extra_context(context_data['extra'])
                
                return func(*args, **kwargs)
                
            except Exception:
                client.captureException()
                raise
            finally:
                # Restore original context
                client.user_context(original_user)
                client.tags_context(original_tags)
                client.extra_context(original_extra)
        
        return wrapper
    return decorator

@with_context(
    tags={'operation': 'payment'},
    extra={'processor': 'stripe'}
)
def process_payment(amount, user_id):
    # Add function-specific context
    client.user_context({'id': user_id})
    client.extra_context({'amount': amount})
    
    # Any errors here will include all context
    return stripe.charge(amount)