Sentry Extension

def get_sentry_extension() -> Callable[[str], None]:
    """
    Return set_transaction_id function if Sentry SDK is installed.
    
    This factory function detects whether the Sentry SDK is available
    and returns the appropriate function for setting correlation IDs
    as Sentry transaction tags.
    
    Returns:
    - Callable[[str], None]: Function to set Sentry transaction ID
    - If Sentry SDK is not installed, returns no-op lambda function
    """

def set_transaction_id(correlation_id: str) -> None:
    """
    Set Sentry's event transaction ID as the current correlation ID.
    
    The transaction ID is displayed in a Sentry event's detail view,
    which makes it easier to correlate logs to specific events.
    
    This function handles different Sentry SDK versions:
    - v2.12.0+: Uses get_isolation_scope() API
    - Earlier versions: Uses configure_scope() context manager
    
    Parameters:
    - correlation_id: The correlation ID to set as transaction tag
    
    Raises:
    - ImportError: If Sentry SDK is not installed (handled internally)
    """

from fastapi import FastAPI
from asgi_correlation_id import CorrelationIdMiddleware
import sentry_sdk

# Initialize Sentry
sentry_sdk.init(
    dsn="your-sentry-dsn",
    traces_sample_rate=1.0,
)

# Add correlation middleware - Sentry integration is automatic
app = FastAPI()
app.add_middleware(CorrelationIdMiddleware)

@app.get("/error")
async def trigger_error():
    # This error will be tagged with the correlation ID in Sentry
    raise ValueError("Test error")

from asgi_correlation_id import correlation_id
from asgi_correlation_id.extensions.sentry import set_transaction_id
import logging

logger = logging.getLogger(__name__)

async def process_request():
    # Get current correlation ID
    current_id = correlation_id.get()
    
    if current_id:
        # Manually tag Sentry transaction
        set_transaction_id(current_id)
        logger.info(f"Processing request {current_id}")
    
    try:
        # Some processing that might fail
        result = risky_operation()
        return result
    except Exception as e:
        # Error will be correlated with logs via shared correlation ID
        logger.error(f"Processing failed: {e}")
        raise

import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration
from asgi_correlation_id import CorrelationIdFilter

# Configure Sentry with logging integration
sentry_logging = LoggingIntegration(
    level=logging.INFO,        # Capture info and above as breadcrumbs
    event_level=logging.ERROR  # Send errors as events
)

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[sentry_logging],
    traces_sample_rate=1.0,
)

# Add correlation ID to logs
correlation_filter = CorrelationIdFilter()
logging.getLogger().addFilter(correlation_filter)

# Now both logs and Sentry events will have correlation IDs

from celery import Celery
from asgi_correlation_id.extensions.celery import load_correlation_ids
from asgi_correlation_id.extensions.sentry import get_sentry_extension
import sentry_sdk

# Initialize Sentry for Celery workers
sentry_sdk.init(dsn="your-sentry-dsn")

app = Celery('tasks')

# Enable correlation ID transfer to Celery tasks
load_correlation_ids()

@app.task
def process_data(data):
    # Correlation ID is automatically transferred from request
    # and tagged in Sentry via the middleware integration
    try:
        return process(data)
    except Exception as e:
        # Error in Sentry will show correlation ID from originating request
        raise

# In CorrelationIdMiddleware.__post_init__()
self.sentry_extension = get_sentry_extension()

# In CorrelationIdMiddleware.__call__()
correlation_id.set(id_value)
self.sentry_extension(id_value)  # Automatically tag Sentry events

# In Celery signal handlers
from asgi_correlation_id.extensions.sentry import get_sentry_extension

sentry_extension = get_sentry_extension()

@task_prerun.connect(weak=False)
def load_correlation_id(task, **kwargs):
    id_value = task.request.get(header_key)
    if id_value:
        correlation_id.set(id_value)
        sentry_extension(id_value)  # Tag Celery task events

Event ID: abc123def456
Transaction ID: req-a1b2c3d4-e5f6-7890  # Your correlation ID

Breadcrumbs:
- INFO: Processing request req-a1b2c3d4-e5f6-7890
- ERROR: Database connection failed

Tags:
- transaction_id: req-a1b2c3d4-e5f6-7890
- environment: production

# Application logs
2023-01-01 12:00:00 INFO [req-a1b2c3d4] Processing user request
2023-01-01 12:00:01 ERROR [req-a1b2c3d4] Database connection failed

# Sentry event tagged with: transaction_id=req-a1b2c3d4
# Easy to correlate the Sentry error with application logs

# Uses newer isolation scope API
scope = sentry_sdk.get_isolation_scope()
scope.set_tag('transaction_id', correlation_id)

# Uses legacy configure_scope API
with sentry_sdk.configure_scope() as scope:
    scope.set_tag('transaction_id', correlation_id)

from typing import Callable
from packaging import version

# Type definitions
SentryExtension = Callable[[str], None]
NoOpExtension = Callable[[str], None]