Tracer

TracerProtocol

Bases: Protocol

Protocol that defines the tracer interface for type checking.

NoOpSpan

A no-op span that does nothing when tracing is disabled.

NoOpTracer

A no-op tracer that returns no-op spans when tracing is disabled.

SpanContextManagerWrapper

SpanContextManagerWrapper(
    context_manager: Any,
    session_id: str | None,
    client_id: str | None = None,
    email: str | None = None,
)

Wrapper for span context managers to add session_id on enter.

Source code in src/rapidata/rapidata_client/config/tracer.py

def __init__(
    self,
    context_manager: Any,
    session_id: str | None,
    client_id: str | None = None,
    email: str | None = None,
):
    self._context_manager = context_manager
    self.session_id = session_id
    self.client_id = client_id
    self.email = email

RapidataTracer

RapidataTracer(name: str = __name__)

Tracer implementation that updates when the configuration changes.

Source code in src/rapidata/rapidata_client/config/tracer.py

def __init__(self, name: str = __name__):
    self._name = name
    self._otlp_initialized = False
    self._init_lock = threading.Lock()
    self._tracer_provider = None
    self._real_tracer = None
    self._no_op_tracer = NoOpTracer()
    self._enabled = True  # Default to enabled
    self._environment = "rapidata.ai"
    self.session_id: str | None = None
    self.client_id: str | None = None
    self.email: str | None = None

    # Register this tracer to receive configuration updates
    register_config_handler(self._handle_config_update)

start_span

start_span(name: str, *args, **kwargs) -> Any

Start a span, or return a no-op span if tracing is disabled.

Source code in src/rapidata/rapidata_client/config/tracer.py

def start_span(self, name: str, *args, **kwargs) -> Any:
    """Start a span, or return a no-op span if tracing is disabled."""
    if self._enabled:
        self._ensure_initialized()
        if self._real_tracer:
            span = self._real_tracer.start_span(name, *args, **kwargs)
            return self._add_attributes_to_span(span)
    return self._no_op_tracer.start_span(name, *args, **kwargs)

start_as_current_span

start_as_current_span(name: str, *args, **kwargs) -> Any

Start a span as current, or return a no-op span if tracing is disabled.

Source code in src/rapidata/rapidata_client/config/tracer.py

def start_as_current_span(self, name: str, *args, **kwargs) -> Any:
    """Start a span as current, or return a no-op span if tracing is disabled."""
    if self._enabled:
        self._ensure_initialized()
        if self._real_tracer:
            context_manager = self._real_tracer.start_as_current_span(
                name, *args, **kwargs
            )
            return SpanContextManagerWrapper(
                context_manager, self.session_id, self.client_id, self.email
            )
    return self._no_op_tracer.start_as_current_span(name, *args, **kwargs)

fail_current_span

fail_current_span(message: str | None = None) -> None

Mark the current span as errored.

Source code in src/rapidata/rapidata_client/config/tracer.py

def fail_current_span(self, message: str | None = None) -> None:
    """Mark the current span as errored."""
    span = trace.get_current_span()
    if span.is_recording():
        span.set_status(Status(StatusCode.ERROR, message))

get_system_attributes

get_system_attributes() -> (
    dict[str, str | int | bool | None]
)

Gather system telemetry for traces.

Source code in src/rapidata/rapidata_client/config/tracer.py

def get_system_attributes() -> dict[str, str | int | bool | None]:
    """Gather system telemetry for traces."""
    try:
        attrs: dict[str, str | int | bool | None] = {
            "system.os": platform.system(),
            "system.os.version": platform.release(),
            "system.arch": platform.machine(),
            "python.version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
            "process.cpu_count": os.cpu_count(),
        }
        agent = detected_coding_agent()
        attrs["agent.detected"] = agent is not None
        if agent:
            attrs["agent.name"] = agent
        logger.debug(f"System attributes: {attrs}")
        return attrs
    except Exception:
        logger.debug("Failed to get system attributes, returning empty dict")
        return {}