OtelSpan

__init__(tracer, request) -> None

def __init__(self, tracer: trace.Tracer, request: SpanRequest) -> None:
    self.name: str = request.name
    self.annotations: Context = request.annotations
    self.links: list[SpanLink] = request.links
    self.kind: SpanKind = request.kind
    active = otel_context.get_current()
    from_active = not request.root and request.parent is None
    self.parent: AnySpan | None = (
        _otel_parent(active, request.annotations) if from_active else request.parent
    )
    if self.parent is None:
        parent_context = trace.set_span_in_context(INVALID_SPAN, active)
    elif from_active and not _overrides_trace_context(request.annotations):
        # Keep the active span object: some SDKs record children on it.
        parent_context = active
    else:
        parent_context = populate_context(active, self.parent, request.annotations)
    self.span: trace.Span = tracer.start_span(
        request.name,
        context=parent_context,
        kind=_KINDS[request.kind],
        links=[_link(link) for link in request.links] or None,
        start_time=request.start_time,
    )
    span_context = self.span.get_span_context()
    self.span_id: str = format_span_id(span_context.span_id)
    self.trace_id: str = format_trace_id(span_context.trace_id)
    self.sampled: bool = is_sampled(span_context.trace_flags)
    self.status: SpanStatus = Started(request.start_time)
    self._attributes: dict[str, object] = {}

attribute(key, value) -> None

def attribute(self, key: str, value: object, /) -> None:
    if isinstance(self.status, Started):
        self.span.set_attribute(key, to_attribute_value(value))
        self._attributes[key] = value

add_links(links) -> None

def add_links(self, links: Sequence[SpanLink], /) -> None:
    if isinstance(self.status, Started):
        self.links.extend(links)
        for link in links:
            self.span.add_link(make_span_context(link.span), to_attributes(link.attributes))

event(name, start_time, attributes=None) -> None

def event(
    self, name: str, start_time: int, attributes: Mapping[str, object] | None = None, /
) -> None:
    if isinstance(self.status, Started):
        self.span.add_event(
            name, None if attributes is None else to_attributes(attributes), start_time
        )

end(end_time, exit) -> None

def end(self, end_time: int, exit: Exit[object, object], /) -> None:
    if isinstance(self.status, Ended):
        return
    self.status = Ended(self.status.start_time, end_time, exit)
    if isinstance(exit, Success):
        self.span.set_status(Status(StatusCode.OK))
    elif exit.cause.has_interrupts_only:
        self.span.set_attribute("effect.fiber.interrupted", True)
    else:
        errors = exit.cause.pretty_errors()
        for error in errors:
            self.span.record_exception(error, timestamp=end_time)
        if errors:
            self.span.set_status(Status(StatusCode.ERROR, str(errors[0])))
        else:
            self.span.set_status(Status(StatusCode.OK))
    self.span.end(end_time)