logfire
logfire copied to clipboard
Improvement to `capfire`
I found this useful, perhaps we should add it to capfire?
class TraceSummary(TypedDict):
id: int
message: str
children: NotRequired[list[TraceSummary]]
@dataclass(init=False)
class LogfireSummary:
traces: list[TraceSummary]
attributes: dict[int, dict[str, Any]]
def __init__(self, capfire: CaptureLogfire):
spans = capfire.exporter.exported_spans_as_dict()
spans.sort(key=lambda s: s['start_time'])
self.traces = []
span_lookup: dict[tuple[str, str], TraceSummary] = {}
self.attributes = {}
id_counter = 0
for span in spans:
tid = span['context']['trace_id'], span['context']['span_id']
span_lookup[tid] = span_summary = TraceSummary(id=id_counter, message=span['attributes']['logfire.msg'])
self.attributes[id_counter] = span['attributes']
id_counter += 1
if parent := span['parent']:
parent_span = span_lookup[(parent['trace_id'], parent['span_id'])]
parent_span.setdefault('children', []).append(span_summary)
else:
self.traces.append(span_summary)
@pytest.fixture
def get_logfire_summary(capfire: CaptureLogfire) -> Callable[[], LogfireSummary]:
def get_summary() -> LogfireSummary:
return LogfireSummary(capfire)
return get_summary