tessl/pypi-ddtrace

Datadog APM client library providing distributed tracing, continuous profiling, error tracking, test optimization, deployment tracking, code hotspots analysis, and dynamic instrumentation for Python applications.

—

Pending

Overview

Eval results

Files

Core Tracing

Name: tessl/pypi-ddtrace
Author: tessl

The core tracing functionality provides the fundamental building blocks for distributed tracing in ddtrace. This includes span creation, context management, trace filtering, and integration configuration that forms the foundation of all observability features.

Capabilities

Tracer Operations

The main Tracer class provides the primary interface for creating spans and managing trace state. It handles span lifecycle, context propagation, and trace submission to the Datadog backend.

class Tracer:
    def trace(
        self,
        name: str,
        service: str = None,
        resource: str = None,
        span_type: str = None
    ) -> Span:
        """
        Create a new span and return it as a context manager.

        Parameters:
        - name: The span name/operation name
        - service: Service name override (defaults to global service)
        - resource: Resource being operated on (e.g., SQL query, HTTP endpoint)
        - span_type: Type of span (web, db, cache, custom, etc.)

        Returns:
        Span object that can be used as a context manager
        """

    def wrap(
        self,
        name: str = None,
        service: str = None,
        resource: str = None,
        span_type: str = None
    ):
        """
        Decorator for wrapping functions with tracing.

        Parameters:
        - name: Span name (defaults to function name)
        - service: Service name override
        - resource: Resource name
        - span_type: Type of span

        Returns:
        Decorator function
        """

Usage examples:

from ddtrace import tracer

# Context manager usage (recommended)
with tracer.trace("database-query") as span:
    span.set_tag("db.statement", "SELECT * FROM users")
    span.set_tag("db.type", "postgresql")
    result = execute_query()
    span.set_tag("db.rows", len(result))

# Manual span management
span = tracer.start_span("background-task")
try:
    span.set_tag("task.type", "data-processing")
    process_data()
except Exception as e:
    span.set_error(e)
finally:
    span.finish()

# Decorator usage
@tracer.wrap("user-validation", service="auth-service")
def validate_user(user_id):
    return check_user_permissions(user_id)

Span Operations

Individual spans represent units of work in a distributed trace. They capture timing information, metadata through tags, and error states.

class Span:
    def set_tag(self, key: str, value: str) -> None:
        """
        Set a tag on the span.

        Parameters:
        - key: Tag key
        - value: Tag value (will be converted to string)
        """

    def set_tags(self, tags: Dict[str, str]) -> None:
        """
        Set multiple tags at once.

        Parameters:
        - tags: Dictionary of key-value pairs
        """

    def set_metric(self, key: str, value: float) -> None:
        """
        Set a numeric metric on the span.

        Parameters:
        - key: Metric key
        - value: Numeric value
        """

    def set_error(self, error: Exception = None, traceback: str = None) -> None:
        """
        Mark the span as having an error.

        Parameters:
        - error: Exception object (optional)
        - traceback: Custom traceback string (optional)
        """

    def finish(self, finish_time: float = None) -> None:
        """
        Finish the span and record its completion time.

        Parameters:
        - finish_time: Custom finish time (defaults to current time)
        """

    @property
    def trace_id(self) -> int:
        """Get the trace ID."""

    @property
    def span_id(self) -> int:
        """Get the span ID."""

    @property
    def parent_id(self) -> int:
        """Get the parent span ID."""

Context Management

Context objects manage the active span state and enable trace propagation across execution contexts.

class Context:
    def clone(self) -> 'Context':
        """
        Create a copy of the context.

        Returns:
        New Context object with the same state
        """

    @property
    def trace_id(self) -> int:
        """Get the trace ID from context."""

    @property
    def span_id(self) -> int:
        """Get the current span ID from context."""

Pin Configuration

Pin objects provide a way to configure tracing behavior for specific objects or modules, commonly used in automatic instrumentation.

class Pin:
    def __init__(
        self,
        service: str = None,
        app: str = None,
        tags: Dict[str, str] = None,
        tracer: Tracer = None
    ):
        """
        Create a Pin configuration object.

        Parameters:
        - service: Service name
        - app: Application name
        - tags: Default tags to apply
        - tracer: Tracer instance to use
        """

    def onto(self, obj: object) -> 'Pin':
        """
        Attach this Pin to an object.

        Parameters:
        - obj: Object to attach Pin to

        Returns:
        The Pin instance for chaining
        """

    @classmethod
    def get_from(cls, obj: object) -> 'Pin':
        """
        Get Pin configuration from an object.

        Parameters:
        - obj: Object to get Pin from

        Returns:
        Pin object or None if not found
        """

    def clone(self, **kwargs) -> 'Pin':
        """
        Create a copy of this Pin with optional overrides.

        Parameters:
        - **kwargs: Attributes to override

        Returns:
        New Pin object
        """

Trace Filtering

TraceFilter enables custom filtering and modification of complete traces before they are submitted.

class TraceFilter:
    def process_trace(self, trace: List[Span]) -> List[Span]:
        """
        Process a complete trace and return the filtered trace.

        Parameters:
        - trace: List of spans in the trace

        Returns:
        List of spans (potentially modified or filtered)
        """

Context Providers

Context providers manage span activation and storage across different execution contexts (threads, async tasks, etc.).

class BaseContextProvider:
    def activate(self, span: Span) -> object:
        """
        Activate a span in the current context.

        Parameters:
        - span: Span to activate

        Returns:
        Token or context object for deactivation
        """

    def active(self) -> Span:
        """
        Get the currently active span.

        Returns:
        Active Span or None
        """

Global Tracer Instance

ddtrace provides a global tracer instance that is automatically configured and ready to use.

tracer: Tracer  # Global tracer instance

Usage:

from ddtrace import tracer

# Use the global tracer directly
with tracer.trace("operation") as span:
    span.set_tag("component", "business-logic")
    do_work()

Manual Instrumentation Helpers

Helper functions for manually instrumenting specific libraries or code sections.

def patch(**patch_modules: bool) -> None:
    """
    Manually patch specific modules for automatic instrumentation.

    Parameters:
    - **patch_modules: Boolean flags for each module to patch
    - raise_errors: Whether to raise errors if patching fails (default: True)

    Example:
    patch(redis=True, psycopg=True, requests=False)
    """

def patch_all(**patch_modules: bool) -> None:
    """
    Automatically patch all supported modules (deprecated).

    Parameters:
    - **patch_modules: Override flags for specific modules

    Note: This function is deprecated in favor of patch() and DD_PATCH_MODULES
    """

Error Handling

Spans can capture error information automatically or manually:

# Automatic error capture with context manager
with tracer.trace("risky-operation") as span:
    span.set_tag("operation.type", "file-processing")
    # If this raises an exception, it's automatically captured
    process_file(filename)

# Manual error capture
span = tracer.start_span("custom-operation")
try:
    risky_operation()
except ValueError as e:
    span.set_error(e)  # Captures exception details
    span.set_tag("error.handled", "true")
    handle_error(e)
except Exception as e:
    span.set_error(e)
    span.set_tag("error.handled", "false")
    raise
finally:
    span.finish()

Advanced Usage

Custom Service Configuration

from ddtrace import tracer

# Configure service information
tracer.set_service_info(
    service="payment-processor",
    app="ecommerce-backend",
    app_type="web"
)

# Per-span service override
with tracer.trace("external-api-call", service="third-party-service") as span:
    response = call_external_api()
    span.set_tag("http.status_code", response.status_code)

Complex Trace Relationships

# Create parent span
with tracer.trace("batch-processing") as parent_span:
    parent_span.set_tag("batch.size", len(items))
    
    for i, item in enumerate(items):
        # Child spans inherit context automatically
        with tracer.trace("process-item") as child_span:
            child_span.set_tag("item.id", item.id)
            child_span.set_tag("item.index", i)
            process_single_item(item)
    
    parent_span.set_tag("batch.status", "completed")

Install with Tessl CLI