Skip to content

Traces#

ragbits.core.audit.traces.set_trace_handlers #

set_trace_handlers(handlers: Handler | list[Handler]) -> None

Set the global trace handlers.

PARAMETER DESCRIPTION
handlers

List of trace handlers to be used.

TYPE: Handler | list[Handler]

RAISES DESCRIPTION
ValueError

If handler is not found.

TypeError

If handler type is invalid.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/__init__.py
def set_trace_handlers(handlers: Handler | list[Handler]) -> None:
    """
    Set the global trace handlers.

    Args:
        handlers: List of trace handlers to be used.

    Raises:
        ValueError: If handler is not found.
        TypeError: If handler type is invalid.
    """
    global _trace_handlers  # noqa: PLW0602

    if isinstance(handlers, Handler):
        handlers = [handlers]

    for handler in handlers:  # type: ignore
        if isinstance(handler, TraceHandler):
            _trace_handlers.append(handler)
        elif isinstance(handler, str):
            if handler == "otel":
                from ragbits.core.audit.traces.otel import OtelTraceHandler

                if not any(isinstance(item, OtelTraceHandler) for item in _trace_handlers):
                    _trace_handlers.append(OtelTraceHandler())
            elif handler == "cli":
                from ragbits.core.audit.traces.cli import CLITraceHandler

                if not any(isinstance(item, CLITraceHandler) for item in _trace_handlers):
                    _trace_handlers.append(CLITraceHandler())
            else:
                raise ValueError(f"Handler {handler} not found.")
        else:
            raise TypeError(f"Invalid handler type: {type(handler)}")

ragbits.core.audit.traces.clear_trace_handlers #

clear_trace_handlers() -> None

Clear all trace handlers.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/__init__.py
def clear_trace_handlers() -> None:
    """
    Clear all trace handlers.
    """
    global _trace_handlers  # noqa: PLW0602
    _trace_handlers.clear()

ragbits.core.audit.traces.trace #

trace(name: str | None = None, **inputs: Any) -> Iterator[SimpleNamespace]

Context manager for processing a trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str | None DEFAULT: None

inputs

The input data.

TYPE: Any DEFAULT: {}

YIELDS DESCRIPTION
SimpleNamespace

The output data.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/__init__.py
@contextmanager
def trace(name: str | None = None, **inputs: Any) -> Iterator[SimpleNamespace]:  # noqa: ANN401
    """
    Context manager for processing a trace.

    Args:
        name: The name of the trace.
        inputs: The input data.

    Yields:
        The output data.
    """
    # We need to go up 2 frames (trace() and __enter__()) to get the parent function.
    parent_frame = inspect.stack()[2].frame
    name = (
        (
            f"{cls.__class__.__qualname__}.{parent_frame.f_code.co_name}"
            if (cls := parent_frame.f_locals.get("self"))
            else parent_frame.f_code.co_name
        )
        if name is None
        else name
    )

    with ExitStack() as stack:
        outputs = [stack.enter_context(handler.trace(name, **inputs)) for handler in _trace_handlers]
        yield (out := SimpleNamespace())
        for output in outputs:
            output.__dict__.update(vars(out))

ragbits.core.audit.traces.traceable #

traceable(func: Callable[P, R]) -> Callable[P, R]

Decorator for making a function traceable.

PARAMETER DESCRIPTION
func

The function to be decorated.

TYPE: Callable[P, R]

RETURNS DESCRIPTION
Callable[P, R]

The decorated function.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/__init__.py
def traceable(func: Callable[P, R]) -> Callable[P, R]:
    """
    Decorator for making a function traceable.

    Args:
        func: The function to be decorated.

    Returns:
        The decorated function.
    """

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        inputs = _get_function_inputs(func, args, kwargs)
        with trace(name=func.__qualname__, **inputs) as outputs:
            returned = func(*args, **kwargs)
            if returned is not None:
                outputs.returned = returned
        return returned

    @wraps(func)
    async def wrapper_async(*args: P.args, **kwargs: P.kwargs) -> R:
        inputs = _get_function_inputs(func, args, kwargs)
        with trace(name=func.__qualname__, **inputs) as outputs:
            returned = await func(*args, **kwargs)  # type: ignore
            if returned is not None:
                outputs.returned = returned
        return returned

    return wrapper_async if asyncio.iscoroutinefunction(func) else wrapper  # type: ignore

ragbits.core.audit.traces.base.TraceHandler #

TraceHandler()

Bases: Generic[SpanT], ABC

Base class for all trace handlers.

Initialize the TraceHandler instance.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
def __init__(self) -> None:
    """
    Initialize the TraceHandler instance.
    """
    super().__init__()
    self._spans = ContextVar[list[SpanT]]("_spans")
    self._spans.set([])

start abstractmethod #

start(name: str, inputs: dict, current_span: SpanT | None = None) -> SpanT

Log input data at the beginning of the trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: dict

current_span

The current trace span.

TYPE: SpanT | None DEFAULT: None

RETURNS DESCRIPTION
SpanT

The updated current trace span.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@abstractmethod
def start(self, name: str, inputs: dict, current_span: SpanT | None = None) -> SpanT:
    """
    Log input data at the beginning of the trace.

    Args:
        name: The name of the trace.
        inputs: The input data.
        current_span: The current trace span.

    Returns:
        The updated current trace span.
    """

stop abstractmethod #

stop(outputs: dict, current_span: SpanT) -> None

Log output data at the end of the trace.

PARAMETER DESCRIPTION
outputs

The output data.

TYPE: dict

current_span

The current trace span.

TYPE: SpanT

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@abstractmethod
def stop(self, outputs: dict, current_span: SpanT) -> None:
    """
    Log output data at the end of the trace.

    Args:
        outputs: The output data.
        current_span: The current trace span.
    """

error abstractmethod #

error(error: Exception, current_span: SpanT) -> None

Log error during the trace.

PARAMETER DESCRIPTION
error

The error that occurred.

TYPE: Exception

current_span

The current trace span.

TYPE: SpanT

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@abstractmethod
def error(self, error: Exception, current_span: SpanT) -> None:
    """
    Log error during the trace.

    Args:
        error: The error that occurred.
        current_span: The current trace span.
    """

trace #

trace(name: str, **inputs: Any) -> Iterator[SimpleNamespace]

Context manager for processing a trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: Any DEFAULT: {}

YIELDS DESCRIPTION
SimpleNamespace

The output data.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@contextmanager
def trace(self, name: str, **inputs: Any) -> Iterator[SimpleNamespace]:  # noqa: ANN401
    """
    Context manager for processing a trace.

    Args:
        name: The name of the trace.
        inputs: The input data.

    Yields:
        The output data.
    """
    self._spans.set(self._spans.get([])[:])
    current_span = self._spans.get()[-1] if self._spans.get() else None

    span = self.start(
        name=name,
        inputs=inputs,
        current_span=current_span,
    )
    self._spans.get().append(span)

    try:
        yield (outputs := SimpleNamespace())
    except Exception as exc:
        span = self._spans.get().pop()
        self.error(error=exc, current_span=span)
        raise exc

    span = self._spans.get().pop()
    self.stop(outputs=vars(outputs), current_span=span)

ragbits.core.audit.traces.cli.CLITraceHandler #

CLITraceHandler()

Bases: TraceHandler[CLISpan]

CLI trace handler.

Initialize the CLITraceHandler instance.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/cli.py
def __init__(self) -> None:
    """
    Initialize the CLITraceHandler instance.
    """
    super().__init__()
    self.live = Live(auto_refresh=False, vertical_overflow="visible")

live instance-attribute #

live = Live(auto_refresh=False, vertical_overflow='visible')

trace #

trace(name: str, **inputs: Any) -> Iterator[SimpleNamespace]

Context manager for processing a trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: Any DEFAULT: {}

YIELDS DESCRIPTION
SimpleNamespace

The output data.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@contextmanager
def trace(self, name: str, **inputs: Any) -> Iterator[SimpleNamespace]:  # noqa: ANN401
    """
    Context manager for processing a trace.

    Args:
        name: The name of the trace.
        inputs: The input data.

    Yields:
        The output data.
    """
    self._spans.set(self._spans.get([])[:])
    current_span = self._spans.get()[-1] if self._spans.get() else None

    span = self.start(
        name=name,
        inputs=inputs,
        current_span=current_span,
    )
    self._spans.get().append(span)

    try:
        yield (outputs := SimpleNamespace())
    except Exception as exc:
        span = self._spans.get().pop()
        self.error(error=exc, current_span=span)
        raise exc

    span = self._spans.get().pop()
    self.stop(outputs=vars(outputs), current_span=span)

start #

start(name: str, inputs: dict, current_span: CLISpan | None = None) -> CLISpan

Log input data at the beginning of the trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: dict

current_span

The current trace span.

TYPE: CLISpan | None DEFAULT: None

RETURNS DESCRIPTION
CLISpan

The updated current trace span.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/cli.py
def start(self, name: str, inputs: dict, current_span: CLISpan | None = None) -> CLISpan:
    """
    Log input data at the beginning of the trace.

    Args:
        name: The name of the trace.
        inputs: The input data.
        current_span: The current trace span.

    Returns:
        The updated current trace span.
    """
    formatter = AttributeFormatter(data=inputs, prefix="inputs")
    formatter.process_attributes()
    attributes = formatter.flattened

    span = CLISpan(
        name=name,
        attributes=attributes,
        parent=current_span,
    )
    if current_span is None:
        self.live = Live(auto_refresh=False, vertical_overflow="visible")
        self.live.start()
        self.tree = span.tree

    span.update()
    self.live.update(self.tree, refresh=True)

    return span

stop #

stop(outputs: dict, current_span: CLISpan) -> None

Log output data at the end of the trace.

PARAMETER DESCRIPTION
outputs

The output data.

TYPE: dict

current_span

The current trace span.

TYPE: CLISpan

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/cli.py
def stop(self, outputs: dict, current_span: CLISpan) -> None:
    """
    Log output data at the end of the trace.

    Args:
        outputs: The output data.
        current_span: The current trace span.
    """
    formatter = AttributeFormatter(data=outputs, prefix="outputs")
    formatter.process_attributes()
    attributes = formatter.flattened
    current_span.attributes.update(attributes)
    current_span.status = SpanStatus.COMPLETED
    current_span.end()

    current_span.update()
    self.live.update(self.tree, refresh=True)

    if current_span.parent is None:
        self.live.stop()

error #

error(error: Exception, current_span: CLISpan) -> None

Log error during the trace.

PARAMETER DESCRIPTION
error

The error that occurred.

TYPE: Exception

current_span

The current trace span.

TYPE: CLISpan

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/cli.py
def error(self, error: Exception, current_span: CLISpan) -> None:
    """
    Log error during the trace.

    Args:
        error: The error that occurred.
        current_span: The current trace span.
    """
    formatter = AttributeFormatter({"message": str(error), **vars(error)}, prefix="error")
    formatter.process_attributes()
    attributes = formatter.flattened
    current_span.attributes.update(attributes)
    current_span.status = SpanStatus.ERROR
    current_span.end()

    current_span.update()
    self.live.update(self.tree, refresh=True)

    if current_span.parent is None:
        self.live.stop()

ragbits.core.audit.traces.otel.OtelTraceHandler #

OtelTraceHandler(provider: TracerProvider | None = None)

Bases: TraceHandler[Span]

OpenTelemetry trace handler.

Initialize the OtelTraceHandler instance.

PARAMETER DESCRIPTION
provider

The tracer provider to use.

TYPE: TracerProvider | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/otel.py
def __init__(self, provider: TracerProvider | None = None) -> None:
    """
    Initialize the OtelTraceHandler instance.

    Args:
        provider: The tracer provider to use.
    """
    super().__init__()
    self._tracer = get_tracer(instrumenting_module_name=__name__, tracer_provider=provider)

trace #

trace(name: str, **inputs: Any) -> Iterator[SimpleNamespace]

Context manager for processing a trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: Any DEFAULT: {}

YIELDS DESCRIPTION
SimpleNamespace

The output data.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/base.py
@contextmanager
def trace(self, name: str, **inputs: Any) -> Iterator[SimpleNamespace]:  # noqa: ANN401
    """
    Context manager for processing a trace.

    Args:
        name: The name of the trace.
        inputs: The input data.

    Yields:
        The output data.
    """
    self._spans.set(self._spans.get([])[:])
    current_span = self._spans.get()[-1] if self._spans.get() else None

    span = self.start(
        name=name,
        inputs=inputs,
        current_span=current_span,
    )
    self._spans.get().append(span)

    try:
        yield (outputs := SimpleNamespace())
    except Exception as exc:
        span = self._spans.get().pop()
        self.error(error=exc, current_span=span)
        raise exc

    span = self._spans.get().pop()
    self.stop(outputs=vars(outputs), current_span=span)

start #

start(name: str, inputs: dict, current_span: Span | None = None) -> Span

Log input data at the beginning of the trace.

PARAMETER DESCRIPTION
name

The name of the trace.

TYPE: str

inputs

The input data.

TYPE: dict

current_span

The current trace span.

TYPE: Span | None DEFAULT: None

RETURNS DESCRIPTION
Span

The updated current trace span.

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/otel.py
def start(self, name: str, inputs: dict, current_span: Span | None = None) -> Span:
    """
    Log input data at the beginning of the trace.

    Args:
        name: The name of the trace.
        inputs: The input data.
        current_span: The current trace span.

    Returns:
        The updated current trace span.
    """
    context = set_span_in_context(current_span) if current_span else None
    with self._tracer.start_as_current_span(name, context=context, end_on_exit=False) as span:
        attributes = format_attributes(inputs, prefix="inputs")
        span.set_attributes(attributes)
    return span

stop #

stop(outputs: dict, current_span: Span) -> None

Log output data at the end of the trace.

PARAMETER DESCRIPTION
outputs

The output data.

TYPE: dict

current_span

The current trace span.

TYPE: Span

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/otel.py
def stop(self, outputs: dict, current_span: Span) -> None:  # noqa: PLR6301
    """
    Log output data at the end of the trace.

    Args:
        outputs: The output data.
        current_span: The current trace span.
    """
    attributes = format_attributes(outputs, prefix="outputs")
    current_span.set_attributes(attributes)
    current_span.set_status(StatusCode.OK)
    current_span.end()

error #

error(error: Exception, current_span: Span) -> None

Log error during the trace.

PARAMETER DESCRIPTION
error

The error that occurred.

TYPE: Exception

current_span

The current trace span.

TYPE: Span

Source code in packages/ragbits-core/src/ragbits/core/audit/traces/otel.py
def error(self, error: Exception, current_span: Span) -> None:  # noqa: PLR6301
    """
    Log error during the trace.

    Args:
        error: The error that occurred.
        current_span: The current trace span.
    """
    attributes = format_attributes({"message": str(error), **vars(error)}, prefix="error")
    current_span.set_attributes(attributes)
    current_span.set_status(StatusCode.ERROR)
    current_span.end()