tessl/pypi-azure-core

Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Distributed Tracing and Diagnostics

Name: tessl/pypi-azure-core
Author: tessl

Azure Core provides comprehensive distributed tracing capabilities with OpenTelemetry integration, automatic HTTP instrumentation, and flexible span management. The tracing system enables end-to-end observability across Azure SDK operations with W3C Trace Context compliance for interoperability.

Core Components

AbstractSpan

Protocol interface for distributed tracing span implementations with context management and attribute support.

from azure.core.tracing import AbstractSpan, SpanKind
from typing import Optional, Dict, Union, Any, Type, ContextManager, Callable

class AbstractSpan(Protocol[SpanType]):
    def __init__(self, span: Optional[SpanType] = None, name: Optional[str] = None, **kwargs: Any) -> None: ...
    
    # Span lifecycle
    def span(self, name: str = "child_span", **kwargs: Any) -> AbstractSpan[SpanType]: ...
    def start(self) -> None: ...
    def finish(self) -> None: ...
    
    # Context manager support
    def __enter__(self) -> AbstractSpan[SpanType]: ...
    def __exit__(self, exception_type: Optional[Type[BaseException]], 
                exception_value: Optional[BaseException], 
                traceback: Any) -> None: ...
    
    # Span properties
    @property
    def kind(self) -> Optional[SpanKind]: ...
    @kind.setter
    def kind(self, value: SpanKind) -> None: ...
    
    @property
    def span_instance(self) -> SpanType: ...
    
    # Attribute management
    def add_attribute(self, key: str, value: Union[str, int]) -> None: ...
    def set_http_attributes(self, request: Any, response: Optional[Any] = None) -> None: ...
    
    # Trace context operations
    def to_header(self) -> Dict[str, str]: ...
    def get_trace_parent(self) -> str: ...
    
    # Class methods for context management
    @classmethod
    def link(cls, traceparent: str, attributes: Optional[Dict[str, Any]] = None) -> None: ...
    @classmethod
    def link_from_headers(cls, headers: Dict[str, str], attributes: Optional[Dict[str, Any]] = None) -> None: ...
    @classmethod
    def get_current_span(cls) -> SpanType: ...
    @classmethod
    def get_current_tracer(cls) -> Any: ...
    @classmethod
    def set_current_span(cls, span: SpanType) -> None: ...
    @classmethod
    def set_current_tracer(cls, tracer: Any) -> None: ...
    @classmethod
    def change_context(cls, span: SpanType) -> ContextManager[SpanType]: ...
    @classmethod
    def with_current_context(cls, func: Callable) -> Callable: ...

SpanKind

Enumeration for categorizing spans based on their role in distributed traces.

from azure.core.tracing import SpanKind
from enum import Enum

class SpanKind(Enum):
    UNSPECIFIED = 1  # Default span kind
    SERVER = 2       # Handles incoming requests
    CLIENT = 3       # Makes outgoing requests
    PRODUCER = 4     # Initiates operations (messaging)
    CONSUMER = 5     # Processes operations from producers
    INTERNAL = 6     # Internal application operations

TracingOptions

Configuration options for controlling tracing behavior.

from azure.core.tracing import TracingOptions
from typing import TypedDict, Mapping, Union, Sequence

AttributeValue = Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]
Attributes = Mapping[str, AttributeValue]

class TracingOptions(TypedDict, total=False):
    enabled: bool           # Override global tracing setting
    attributes: Attributes  # Additional span attributes

Link

Represents references between spans across service boundaries.

from azure.core.tracing import Link
from typing import Dict, Optional

class Link:
    def __init__(self, headers: Dict[str, str], attributes: Optional[Attributes] = None) -> None: ...
    
    headers: Dict[str, str]
    attributes: Optional[Attributes]

Basic Usage

Manual Span Management

from azure.core.tracing import AbstractSpan, SpanKind

def manual_tracing_example():
    # Create span with context manager
    with AbstractSpan(name="operation", kind=SpanKind.CLIENT) as span:
        span.add_attribute("user.id", "12345")
        span.add_attribute("operation.type", "data_fetch")
        
        try:
            # Your operation code here
            result = perform_operation()
            span.add_attribute("operation.success", True)
            span.add_attribute("result.count", len(result))
            return result
        except Exception as e:
            span.add_attribute("error.type", type(e).__name__)
            span.add_attribute("error.message", str(e))
            raise

def nested_spans_example():
    with AbstractSpan(name="parent_operation", kind=SpanKind.INTERNAL) as parent:
        parent.add_attribute("operation.phase", "initialization")
        
        # Create child span
        with parent.span(name="child_operation", kind=SpanKind.CLIENT) as child:
            child.add_attribute("target.service", "external-api")
            perform_external_call()
        
        parent.add_attribute("operation.phase", "completion")

Decorator-Based Tracing

from azure.core.tracing.decorator import distributed_trace
from azure.core.tracing import SpanKind

@distributed_trace
def simple_traced_function():
    """Automatically traced with function name as span name"""
    return process_data()

@distributed_trace(
    name_of_span="custom_operation_name",
    kind=SpanKind.CLIENT,
    tracing_attributes={"service.name": "my-service", "operation.type": "fetch"}
)
def custom_traced_function(user_id: str):
    """Traced with custom span name, kind, and attributes"""
    return fetch_user_data(user_id)

# Async version
from azure.core.tracing.decorator_async import distributed_trace_async

@distributed_trace_async(name_of_span="async_operation", kind=SpanKind.CLIENT)
async def async_traced_function():
    """Async function with automatic tracing"""
    return await async_operation()

HTTP Request Tracing

from azure.core.tracing import AbstractSpan, SpanKind
from azure.core.rest import HttpRequest

def http_request_tracing():
    request = HttpRequest("GET", "https://api.example.com/data")
    
    with AbstractSpan(name="http_call", kind=SpanKind.CLIENT) as span:
        # Automatically sets HTTP attributes
        span.set_http_attributes(request)
        
        # Make request
        response = client.send_request(request)
        
        # Update span with response information
        span.set_http_attributes(request, response)
        
        return response

OpenTelemetry Integration

OpenTelemetrySpan

Complete AbstractSpan implementation using OpenTelemetry spans.

from azure.core.tracing.ext.opentelemetry_span import OpenTelemetrySpan
from azure.core.tracing import SpanKind

# Basic OpenTelemetry span usage
with OpenTelemetrySpan(name="azure_operation", kind=SpanKind.CLIENT) as span:
    span.add_attribute("azure.namespace", "Microsoft.Storage")
    span.add_attribute("azure.service", "BlobService")
    
    # Your Azure SDK operation
    result = perform_azure_operation()

# With links to parent traces
parent_headers = {"traceparent": "00-trace-id-span-id-01"}
links = [Link(parent_headers, {"link.type": "follows_from"})]

with OpenTelemetrySpan(name="linked_operation", links=links) as span:
    # Operation that follows from another trace
    process_linked_operation()

Context Propagation

from azure.core.tracing import AbstractSpan

def context_propagation_example():
    with AbstractSpan(name="parent") as parent_span:
        # Get trace context headers for propagation
        trace_headers = parent_span.to_header()
        
        # Headers contain W3C Trace Context
        # traceparent: 00-trace-id-span-id-flags
        # tracestate: (optional state information)
        
        # Pass headers to downstream service
        request = HttpRequest("POST", "/downstream", headers=trace_headers)
        response = client.send_request(request)
        
        return response

def incoming_trace_context():
    # Receive headers from upstream service
    incoming_headers = {
        "traceparent": "00-12345678901234567890123456789012-1234567890123456-01"
    }
    
    # Create links to connect traces
    AbstractSpan.link_from_headers(incoming_headers, {"link.type": "follows_from"})

Pipeline Integration

DistributedTracingPolicy

Automatic HTTP request tracing through the pipeline system.

from azure.core.pipeline.policies import DistributedTracingPolicy
from azure.core import PipelineClient

# Pipeline client with automatic tracing
client = PipelineClient(
    base_url="https://api.service.com",
    policies=[
        DistributedTracingPolicy(),
        # ... other policies
    ]
)

# All HTTP requests through this client are automatically traced
request = HttpRequest("GET", "/api/data")
response = client.send_request(request)  # Automatically creates span

Custom Instrumentation Configuration

from azure.core.pipeline.policies import DistributedTracingPolicy

# Custom instrumentation configuration
instrumentation_config = {
    "span_name": "custom_http_call",
    "attributes": {
        "service.name": "my-azure-client",
        "service.version": "1.0.0"
    }
}

policy = DistributedTracingPolicy(instrumentation_config=instrumentation_config)

Advanced Features

Custom Tracer Integration

from azure.core.tracing import AbstractSpan

class CustomTracer:
    def create_span(self, name: str, **kwargs):
        # Your custom span implementation
        return CustomSpan(name, **kwargs)

# Set custom tracer globally
AbstractSpan.set_current_tracer(CustomTracer())

# All spans will now use your custom implementation
with AbstractSpan(name="operation") as span:
    # Uses CustomTracer.create_span
    pass

Span Suppression

from azure.core.settings import settings

# Disable tracing globally
settings.tracing_enabled = False

# Or per operation using TracingOptions
@distributed_trace(tracing_attributes={"enabled": False})
def untraced_operation():
    # This operation won't be traced
    pass

Error Handling and Status

from azure.core.tracing import AbstractSpan, SpanKind

def error_handling_example():
    with AbstractSpan(name="operation_with_errors", kind=SpanKind.INTERNAL) as span:
        try:
            risky_operation()
            span.add_attribute("operation.status", "success")
        except ValueError as e:
            # Automatic error attributes
            span.add_attribute("error.type", "ValueError")
            span.add_attribute("error.message", str(e))
            span.add_attribute("operation.status", "error")
            raise
        except Exception as e:
            span.add_attribute("error.type", type(e).__name__)
            span.add_attribute("error.message", str(e))
            span.add_attribute("operation.status", "error")
            raise

Async Context Preservation

import asyncio
from azure.core.tracing import AbstractSpan
from azure.core.tracing.decorator_async import distributed_trace_async

@distributed_trace_async(name_of_span="async_coordinator")
async def async_coordinator():
    # Context is automatically preserved across await boundaries
    tasks = [
        async_task_1(),
        async_task_2(),
        async_task_3()
    ]
    
    # All tasks will be child spans of the coordinator span
    results = await asyncio.gather(*tasks)
    return results

@distributed_trace_async(name_of_span="async_task", kind=SpanKind.INTERNAL)
async def async_task_1():
    await asyncio.sleep(1)  # Simulated async work
    return "task1_result"

Batch Operations

from azure.core.tracing import AbstractSpan, SpanKind

def batch_operation_tracing():
    with AbstractSpan(name="batch_processor", kind=SpanKind.INTERNAL) as batch_span:
        batch_span.add_attribute("batch.size", 100)
        batch_span.add_attribute("batch.type", "data_processing")
        
        for i, item in enumerate(batch_items):
            # Create span for each item or group of items
            with batch_span.span(name=f"process_item_{i}", kind=SpanKind.INTERNAL) as item_span:
                item_span.add_attribute("item.id", item.id)
                item_span.add_attribute("item.type", item.type)
                
                try:
                    result = process_item(item)
                    item_span.add_attribute("processing.success", True)
                except Exception as e:
                    item_span.add_attribute("processing.success", False)
                    item_span.add_attribute("error.type", type(e).__name__)
                    # Continue processing other items
                    continue

Configuration and Settings

Global Settings

from azure.core.settings import settings

# Enable/disable tracing globally
settings.tracing_enabled = True

# Configure tracing options
settings.tracing_implementation = "OpenTelemetry"  # or custom implementation

Per-Operation Configuration

# Using TracingOptions
tracing_options = {
    "enabled": True,
    "attributes": {
        "user.id": "12345",
        "operation.priority": "high"
    }
}

@distributed_trace(tracing_attributes=tracing_options)
def configured_operation():
    pass

Key Features

OpenTelemetry Integration: Full compatibility with OpenTelemetry standards and tooling.

W3C Trace Context: Automatic propagation of trace context using W3C standards for interoperability.

Automatic HTTP Instrumentation: Pipeline policy automatically traces HTTP requests with semantic conventions.

Flexible Span Management: Support for manual span creation, decorators, and context managers.

Cross-Service Linking: Link spans across service boundaries for complete trace visibility.

Error Handling: Automatic error capture with structured error attributes.

Async Support: Full async/await support with proper context preservation.

Custom Tracer Support: Plugin architecture for custom tracing implementations.

Performance Optimized: Minimal overhead when tracing is disabled, efficient span creation.

The distributed tracing system provides comprehensive observability for Azure SDK operations while maintaining flexibility for custom implementations and seamless integration with existing observability infrastructure.

Install with Tessl CLI