GitHub

EffectSpanView

API reference for EffectSpanView.

Writes through to span: attributes, events, links and the end, with the exit the last set_status chose (an error status ends it with a defect). Times default to clock's now.

Functions

__init__

__init__(span, clock) -> None
ParameterTypeDescription
spanSpan
clockClockService
def __init__(self, span: Span, clock: ClockService) -> None:
    self._span = span
    self._clock = clock
    self._exit: Exit[object, object] = Success(None)
    self._context = SpanContext(
        hex_id(span.trace_id),
        hex_id(span.span_id),
        False,
        TraceFlags(TraceFlags.SAMPLED if span.sampled else TraceFlags.DEFAULT),
    )

get_span_context

get_span_context() -> SpanContext

Returns SpanContext

def get_span_context(self) -> SpanContext:
    return self._context

set_attribute

set_attribute(key, value) -> None
ParameterTypeDescription
keystr
valuetypes.AttributeValue
def set_attribute(self, key: str, value: types.AttributeValue) -> None:
    self._span.attribute(key, value)

set_attributes

set_attributes(attributes) -> None
ParameterTypeDescription
attributesMapping[str, types.AttributeValue]
def set_attributes(self, attributes: Mapping[str, types.AttributeValue]) -> None:
    for key, value in attributes.items():
        self._span.attribute(key, value)

add_event

add_event(name, attributes=None, timestamp=None) -> None
ParameterTypeDescription
namestr
attributestypes.Attributes = None
timestampint | None = None
def add_event(
    self, name: str, attributes: types.Attributes = None, timestamp: int | None = None
) -> None:
    self._span.event(name, self._time(timestamp), attributes)
add_link(context, attributes=None) -> None
ParameterTypeDescription
contextSpanContext
attributestypes.Attributes = None
def add_link(self, context: SpanContext, attributes: types.Attributes = None) -> None:
    self._span.add_links([_link(context, attributes)])

update_name

update_name(name) -> None
ParameterTypeDescription
namestr
def update_name(self, name: str) -> None:
    pass

is_recording

is_recording() -> bool

Returns bool

def is_recording(self) -> bool:
    return True

set_status

set_status(status, description=None) -> None
ParameterTypeDescription
statusStatus | StatusCode
descriptionstr | None = None
def set_status(self, status: Status | StatusCode, description: str | None = None) -> None:
    code = status.status_code if isinstance(status, Status) else status
    if code is StatusCode.ERROR:
        message = status.description if isinstance(status, Status) else description
        self._exit = die(Exception(message or "Unknown error"))
    else:
        self._exit = Success(None)

end

end(end_time=None) -> None
ParameterTypeDescription
end_timeint | None = None
def end(self, end_time: int | None = None) -> None:
    self._span.end(self._time(end_time), self._exit)

record_exception

record_exception(exception, attributes=None, timestamp=None, escaped=False) -> None
ParameterTypeDescription
exceptionBaseException
attributestypes.Attributes = None
timestampint | None = None
escapedbool = False
def record_exception(
    self,
    exception: BaseException,
    attributes: types.Attributes = None,
    timestamp: int | None = None,
    escaped: bool = False,
) -> None:
    self._span.event(
        str(exception),
        self._time(timestamp),
        {
            "exception.type": type(exception).__name__,
            "exception.message": str(exception),
            "exception.stacktrace": "".join(traceback.format_exception(exception)),
        },
    )