def to_dict()

in elasticapm/traces.py [0:0]

• 58 lines of code
• 25 McCabe index (conditional complexity)

    def to_dict(self) -> dict:
        if (
            self.composite
            and self.composite["compression_strategy"] == "same_kind"
            and nested_key(self.context, "destination", "service", "resource")
        ):
            name = "Calls to " + self.context["destination"]["service"]["resource"]
        else:
            name = self.name
        result = {
            "id": self.id,
            "transaction_id": self.transaction.id,
            "trace_id": self.transaction.trace_parent.trace_id,
            # use either the explicitly set parent_span_id, or the id of the parent, or finally the transaction id
            "parent_id": self.parent_span_id or (self.parent.id if self.parent else self.transaction.id),
            "name": encoding.keyword_field(name),
            "type": encoding.keyword_field(self.type),
            "subtype": encoding.keyword_field(self.subtype),
            "action": encoding.keyword_field(self.action),
            "timestamp": int(self.timestamp * 1000000),  # microseconds
            "duration": self.duration.total_seconds() * 1000,
            "outcome": self.outcome,
        }
        if self.transaction.sample_rate is not None:
            result["sample_rate"] = float(self.transaction.sample_rate)
        if self.sync is not None:
            result["sync"] = self.sync
        if self.labels:
            if self.context is None:
                self.context = {}
            self.context["tags"] = self.labels
        if self.links:
            result["links"] = self.links
        if self.context:
            self.autofill_resource_context()
            # otel attributes and spankind need to be top-level
            if "otel_spankind" in self.context:
                result["otel"] = {"span_kind": self.context.pop("otel_spankind")}
            if self.tracer._agent.check_server_version(gte=(7, 16)):
                if "otel_attributes" in self.context:
                    if "otel" not in result:
                        result["otel"] = {"attributes": self.context.pop("otel_attributes")}
                    else:
                        result["otel"]["attributes"] = self.context.pop("otel_attributes")
            else:
                # Attributes map to labels for older versions
                attributes = self.context.pop("otel_attributes", {})
                if attributes and ("tags" not in self.context):
                    self.context["tags"] = {}
                for key, value in attributes.items():
                    self.context["tags"][key] = value
            result["context"] = self.context
        if self.frames:
            result["stacktrace"] = self.frames
        if self.composite:
            result["composite"] = {
                "compression_strategy": self.composite["compression_strategy"],
                "sum": self.composite["sum"].total_seconds() * 1000,
                "count": self.composite["count"],
            }
        return result