Tracer

provide.foundation.tracer

TODO: Add module docstring.

Classes

Span

Enhanced span implementation with optional OpenTelemetry integration.

Maintains simple API while providing distributed tracing when OpenTelemetry is available.

Functions

__attrs_post_init__

__attrs_post_init__() -> None

Initialize span after creation.

Source code in provide/foundation/tracer/spans.py

def __attrs_post_init__(self) -> None:
    """Initialize span after creation."""
    # Set up time source
    self._time_source = self.time_source if self.time_source is not None else time.time

    # Set start_time if not provided
    if self.start_time is None:
        object.__setattr__(self, "start_time", self._time_source())

    # Try to create OpenTelemetry span if available
    if _HAS_OTEL:
        try:
            tracer = otel_trace.get_tracer(__name__)
            self._otel_span = tracer.start_span(self.name)

            log.debug(f"🔍✨ Created OpenTelemetry span: {self.name}")
        except Exception as e:
            log.debug(f"🔍⚠️ Failed to create OpenTelemetry span: {e}")
            self._otel_span = None

__enter__

__enter__() -> Span

Context manager entry.

Source code in provide/foundation/tracer/spans.py

def __enter__(self) -> Span:
    """Context manager entry."""
    # Set this span as current in OpenTelemetry context if available
    if self._otel_span and _HAS_OTEL:
        try:
            # OpenTelemetry spans are automatically set as current when started
            pass
        except Exception as e:
            log.debug(f"🔍⚠️ Failed to set OpenTelemetry span context: {e}")

    # Also set in Foundation tracer context
    try:
        from provide.foundation.tracer.context import set_current_span

        set_current_span(self)
    except Exception as e:
        log.debug(f"🔍⚠️ Failed to set Foundation span context: {e}")

    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None

Context manager exit.

Source code in provide/foundation/tracer/spans.py

def __exit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
) -> None:
    """Context manager exit."""
    # Handle exceptions
    if exc_type is not None:
        error_msg = str(exc_val) if exc_val else exc_type.__name__
        if isinstance(exc_val, Exception):
            self.set_error(exc_val)
        else:
            self.set_error(error_msg)

    # Finish the span
    self.finish()

    # Clear from Foundation tracer context
    try:
        from provide.foundation.tracer.context import set_current_span

        set_current_span(None)
    except Exception as e:
        log.debug(f"🔍⚠️ Failed to clear Foundation span context: {e}")

duration_ms

duration_ms() -> float

Get the duration of the span in milliseconds.

Source code in provide/foundation/tracer/spans.py

def duration_ms(self) -> float:
    """Get the duration of the span in milliseconds."""
    # start_time is guaranteed to be set in __attrs_post_init__
    start = self.start_time if self.start_time is not None else 0.0
    if self.end_time is None:
        return (self._time_source() - start) * 1000
    return (self.end_time - start) * 1000

finish

finish() -> None

Finish the span and record end time.

Source code in provide/foundation/tracer/spans.py

def finish(self) -> None:
    """Finish the span and record end time."""
    if self._active:
        self.end_time = self._time_source()
        self._active = False

        # Also finish OpenTelemetry span if available
        if self._otel_span:
            try:
                self._otel_span.end()
            except Exception as e:
                log.debug(f"🔍⚠️ Failed to finish OpenTelemetry span: {e}")

set_error

set_error(error: str | Exception) -> None

Mark the span as having an error.

Source code in provide/foundation/tracer/spans.py

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

    # Also set on OpenTelemetry span if available
    if self._otel_span and Status is not None and StatusCode is not None:
        try:
            self._otel_span.set_status(Status(StatusCode.ERROR, str(error)))
            self._otel_span.record_exception(error if isinstance(error, Exception) else Exception(error))
        except Exception as e:
            log.debug(f"🔍⚠️ Failed to set OpenTelemetry error: {e}")

set_tag

set_tag(key: str, value: Any) -> None

Set a tag on the span.

Source code in provide/foundation/tracer/spans.py

def set_tag(self, key: str, value: Any) -> None:
    """Set a tag on the span."""
    self.tags[key] = value

    # Also set on OpenTelemetry span if available
    if self._otel_span and hasattr(self._otel_span, "set_attribute"):
        try:
            self._otel_span.set_attribute(key, value)
        except Exception as e:
            log.debug(f"🔍⚠️ Failed to set OpenTelemetry attribute: {e}")

to_dict

to_dict() -> dict[str, Any]

Convert span to dictionary representation.

Source code in provide/foundation/tracer/spans.py

def to_dict(self) -> dict[str, Any]:
    """Convert span to dictionary representation."""
    return {
        "name": self.name,
        "span_id": self.span_id,
        "parent_id": self.parent_id,
        "trace_id": self.trace_id,
        "start_time": self.start_time,
        "end_time": self.end_time,
        "duration_ms": self.duration_ms(),
        "tags": self.tags,
        "status": self.status,
        "error": self.error,
    }

Functions

get_current_span

get_current_span() -> Span | None

Get the currently active span.

Source code in provide/foundation/tracer/context.py

def get_current_span() -> Span | None:
    """Get the currently active span."""
    return _current_span.get(None)

get_current_trace_id

get_current_trace_id() -> str | None

Get the current trace ID.

Source code in provide/foundation/tracer/context.py

def get_current_trace_id() -> str | None:
    """Get the current trace ID."""
    return _current_trace_id.get(None)

get_trace_context

get_trace_context() -> dict[str, Any]

Get the current trace context information.

Returns:

Type	Description
dict[str, Any]	Dictionary with trace context information

Source code in provide/foundation/tracer/context.py

def get_trace_context() -> dict[str, Any]:
    """Get the current trace context information.

    Returns:
        Dictionary with trace context information

    """
    current_span = get_current_span()
    trace_id = get_current_trace_id()

    return {
        "trace_id": trace_id,
        "span_id": current_span.span_id if current_span else None,
        "span_name": current_span.name if current_span else None,
    }

set_current_span

set_current_span(span: Span | None) -> None

Set the current active span.

Source code in provide/foundation/tracer/context.py

def set_current_span(span: Span | None) -> None:
    """Set the current active span."""
    _current_span.set(span)
    if span:
        _current_trace_id.set(span.trace_id)

with_span

with_span(name: str) -> SpanContext

Create a new span context.

Parameters:

Name	Type	Description	Default
name	str	Name of the span	required

Returns:

Type	Description
SpanContext	SpanContext that can be used as a context manager

Source code in provide/foundation/tracer/context.py

def with_span(name: str) -> SpanContext:
    """Create a new span context.

    Args:
        name: Name of the span

    Returns:
        SpanContext that can be used as a context manager

    """
    span = create_child_span(name)
    return SpanContext(span)