def to_dict()

in elasticapm/traces.py [0:0]

• 51 lines of code
• 15 McCabe index (conditional complexity)

    def to_dict(self) -> dict:
        # This copy() only covers top level `.pop()` calls, so if we ever start
        # modifying nested data, we'll need to do a deep copy.
        context = self.context.copy()
        context["tags"] = self.labels
        result = {
            "id": self.id,
            "trace_id": self.trace_parent.trace_id,
            "name": encoding.keyword_field(self.name or ""),
            "type": encoding.keyword_field(self.transaction_type),
            "duration": self.duration.total_seconds() * 1000 if self.duration else None,
            "result": encoding.keyword_field(str(self.result)),
            "timestamp": int(self.timestamp * 1_000_000),  # microseconds
            "outcome": self.outcome,
            "sampled": self.is_sampled,
            "span_count": {"started": self._span_counter, "dropped": self.dropped_spans},
        }
        if self._dropped_span_statistics:
            result["dropped_spans_stats"] = [
                {
                    "destination_service_resource": resource,
                    "service_target_type": target_type,
                    "service_target_name": target_name,
                    "outcome": outcome,
                    "duration": {"count": v["count"], "sum": {"us": int(v["duration.sum.us"])}},
                }
                for (resource, outcome, target_type, target_name), v in self._dropped_span_statistics.items()
            ]
        if self.sample_rate is not None:
            result["sample_rate"] = float(self.sample_rate)
        if self.trace_parent:
            result["trace_id"] = self.trace_parent.trace_id
            # only set parent_id if this transaction isn't the root
            if self.trace_parent.span_id and self.trace_parent.span_id != self.id:
                result["parent_id"] = self.trace_parent.span_id
        if self.links:
            result["links"] = self.links
        # faas context belongs top-level on the transaction
        if "faas" in context:
            result["faas"] = context.pop("faas")
        # otel attributes and spankind need to be top-level
        if "otel_spankind" in context:
            result["otel"] = {"span_kind": context.pop("otel_spankind")}
        # Some transaction_store_tests use the Tracer without a Client -- the
        # extra check against `get_client()` is here to make those tests pass
        if elasticapm.get_client() and elasticapm.get_client().check_server_version(gte=(7, 16)):
            if "otel_attributes" in context:
                if "otel" not in result:
                    result["otel"] = {"attributes": context.pop("otel_attributes")}
                else:
                    result["otel"]["attributes"] = context.pop("otel_attributes")
        else:
            # Attributes map to labels for older versions
            attributes = context.pop("otel_attributes", {})
            for key, value in attributes.items():
                result["context"]["tags"][key] = value
        if self.is_sampled:
            result["context"] = context
        return result