tessl/pypi-opentelemetry-api

OpenTelemetry Python API providing core abstractions for distributed tracing, metrics collection, and context propagation

—

Pending

Overview

Eval results

Files

Context Management

Name: tessl/pypi-opentelemetry-api
Author: tessl

Comprehensive context management system for cross-cutting concerns in distributed applications. OpenTelemetry context provides immutable context propagation with copy-on-write semantics, enabling trace correlation, baggage handling, and instrumentation control across async boundaries and service calls.

Capabilities

Context Creation and Access

Manage immutable context objects for storing and propagating cross-cutting data.

class Context:
    """Immutable context dictionary for cross-cutting concerns."""
    
    def __init__(self, data: Optional[Dict[str, object]] = None) -> None:
        """
        Creates a new context.
        
        Parameters:
        - data: Optional initial context data
        """
    
    def get(self, key: str) -> object:
        """
        Returns the value associated with the key.
        
        Parameters:
        - key: The context key
        
        Returns:
        The value associated with the key, or None if not found
        """
    
    def copy(self) -> Dict[str, object]:
        """
        Returns a shallow copy of the context data.
        
        Returns:
        Dictionary containing context key-value pairs
        """

def get_current() -> Context:
    """
    Returns the current context.
    
    Returns:
    The current Context object
    """

Key Management

Create and manage unique keys for context storage.

def create_key(keyname: str) -> str:
    """
    Creates a unique key for context storage.
    
    Parameters:
    - keyname: The key name for debugging purposes (not required to be unique)
    
    Returns:
    A unique string representing the newly created key
    """

Context Value Operations

Store and retrieve values in context with immutable semantics.

def get_value(key: str, context: Optional[Context] = None) -> object:
    """
    Retrieves a value from the context.
    
    Parameters:
    - key: The key of the value to retrieve
    - context: The context from which to retrieve the value, if None uses current context
    
    Returns:
    The value associated with the key
    """

def set_value(
    key: str, 
    value: object, 
    context: Optional[Context] = None
) -> Context:
    """
    Creates a new context with the specified key-value pair.
    
    Parameters:
    - key: The key of the entry to set
    - value: The value of the entry to set
    - context: The context to copy, if None uses current context
    
    Returns:
    A new Context containing the value set
    """

Context Activation

Attach and detach contexts for scoped execution.

def attach(context: Context) -> Token[Context]:
    """
    Associates a Context with the caller's current execution unit.
    
    Parameters:
    - context: The Context to set as current
    
    Returns:
    A token that can be used with detach to reset the context
    """

def detach(token: Token[Context]) -> None:
    """
    Resets the Context to the value before attaching a specified Context.
    
    Parameters:
    - token: The Token that was returned by a previous call to attach
    """

Baggage Operations

Store and retrieve cross-service data using the baggage system.

def get_all(context: Optional[Context] = None) -> Mapping[str, object]:
    """
    Returns all name/value pairs in the Baggage.
    
    Parameters:
    - context: The Context to use, if not set uses current Context
    
    Returns:
    The name/value pairs in the Baggage
    """

def get_baggage(name: str, context: Optional[Context] = None) -> Optional[object]:
    """
    Returns the value for a name/value pair in the Baggage.
    
    Parameters:
    - name: The name of the value to retrieve
    - context: The Context to use, if not set uses current Context
    
    Returns:
    The value associated with the given name, or None if not present
    """

def set_baggage(
    name: str, 
    value: object, 
    context: Optional[Context] = None
) -> Context:
    """
    Sets a value in the Baggage.
    
    Parameters:
    - name: The name of the value to set
    - value: The value to set
    - context: The Context to use, if not set uses current Context
    
    Returns:
    A Context with the value updated
    """

def remove_baggage(name: str, context: Optional[Context] = None) -> Context:
    """
    Removes a value from the Baggage.
    
    Parameters:
    - name: The name of the value to remove
    - context: The Context to use, if not set uses current Context
    
    Returns:
    A Context with the name/value removed
    """

def clear(context: Optional[Context] = None) -> Context:
    """
    Removes all values from the Baggage.
    
    Parameters:
    - context: The Context to use, if not set uses current Context
    
    Returns:
    A Context with all baggage entries removed
    """

Runtime Context Implementation

Interface for runtime context implementations with different backends.

class _RuntimeContext(ABC):
    """Abstract base class for runtime context implementations."""
    
    def attach(self, context: Context) -> Token[Context]:
        """Attaches the given context and returns a token for detaching."""
    
    def detach(self, token: Token[Context]) -> None:
        """Detaches the context associated with the given token."""
    
    def get_current(self) -> Context:
        """Returns the current context."""

class ContextVarsRuntimeContext(_RuntimeContext):
    """Runtime context implementation using contextvars."""
    
    def attach(self, context: Context) -> Token[Context]:
        """Attaches context using contextvars."""
    
    def detach(self, token: Token[Context]) -> None:
        """Detaches context using contextvars."""
    
    def get_current(self) -> Context:
        """Returns current context from contextvars."""

Usage Examples

Basic Context Operations

from opentelemetry import context

# Create a context key
user_key = context.create_key("user.id")

# Set a value in context
ctx = context.set_value(user_key, "12345")

# Retrieve the value
user_id = context.get_value(user_key, ctx)
print(f"User ID: {user_id}")  # Output: User ID: 12345

# Get current context
current_ctx = context.get_current()

Context Activation with Tokens

from opentelemetry import context

# Create a new context with data
request_key = context.create_key("request.id")
new_ctx = context.set_value(request_key, "req-123")

# Activate the context
token = context.attach(new_ctx)

try:
    # Now this context is active
    request_id = context.get_value(request_key)  # Uses current context
    print(f"Request ID: {request_id}")  # Output: Request ID: req-123
    
    # Do work in this context
    process_request()
    
finally:
    # Always detach to restore previous context
    context.detach(token)

Cross-Service Baggage Propagation

from opentelemetry import baggage, context

# Set baggage values
ctx = baggage.set_baggage("user.id", "12345")
ctx = baggage.set_baggage("session.id", "abc-def", ctx)

# Activate the context
token = context.attach(ctx)

try:
    # Baggage is available in current context
    user_id = baggage.get_baggage("user.id")
    session_id = baggage.get_baggage("session.id")
    
    print(f"User: {user_id}, Session: {session_id}")
    
    # Get all baggage
    all_baggage = baggage.get_all()
    print(f"All baggage: {dict(all_baggage)}")
    
    # Make service call - baggage will be propagated
    call_downstream_service()
    
finally:
    context.detach(token)

Context Manager Pattern

from opentelemetry import context, baggage
from contextlib import contextmanager

@contextmanager
def with_user_context(user_id: str):
    """Context manager for user-scoped operations."""
    ctx = baggage.set_baggage("user.id", user_id)
    token = context.attach(ctx)
    try:
        yield
    finally:
        context.detach(token)

# Usage
with with_user_context("12345"):
    # All operations in this block have user context
    user_id = baggage.get_baggage("user.id")
    perform_user_operation()

Async Context Propagation

import asyncio
from opentelemetry import context, baggage

async def async_operation():
    """Async function that preserves context."""
    # Context is automatically propagated in async calls
    user_id = baggage.get_baggage("user.id")
    print(f"Processing for user: {user_id}")
    
    await asyncio.sleep(1)  # Simulated async work
    
    # Context is still available
    user_id = baggage.get_baggage("user.id")
    print(f"Completed for user: {user_id}")

async def main():
    # Set up context
    ctx = baggage.set_baggage("user.id", "12345")
    token = context.attach(ctx)
    
    try:
        # Start async operations - context propagates automatically
        await async_operation()
        
        # Multiple concurrent operations maintain their context
        await asyncio.gather(
            async_operation(),
            async_operation(),
        )
        
    finally:
        context.detach(token)

# Run async example
asyncio.run(main())

Context Validation and Error Handling

from opentelemetry import context, baggage

def safe_baggage_operations():
    """Demonstrate safe baggage operations."""
    try:
        # Set valid baggage
        ctx = baggage.set_baggage("service.name", "user-service")
        ctx = baggage.set_baggage("service.version", "1.2.3", ctx)
        
        token = context.attach(ctx)
        
        try:
            # Get baggage safely
            service_name = baggage.get_baggage("service.name")
            if service_name:
                print(f"Service: {service_name}")
            
            # Handle missing keys gracefully
            missing_value = baggage.get_baggage("nonexistent.key")
            if missing_value is None:
                print("Key not found, using default")
            
            # Remove specific baggage
            ctx = baggage.remove_baggage("service.version")
            
            # Clear all baggage
            ctx = baggage.clear()
            
        finally:
            context.detach(token)
            
    except Exception as e:
        print(f"Context operation failed: {e}")
        # Handle context errors appropriately

Constants and Internal Keys

# Internal context keys (for reference, not direct usage)
_SUPPRESS_INSTRUMENTATION_KEY: str  # Key for suppressing instrumentation
_SUPPRESS_HTTP_INSTRUMENTATION_KEY: str  # Key for suppressing HTTP instrumentation

Install with Tessl CLI

npx tessl i tessl/pypi-opentelemetry-api

docs