How to Model Span Events in Python

Define a Span class with timestamped milestone events and a completion marker to track operation lifecycle.

Easy Python 3.9+ Aug 9, 2026 Observability & SRE 14 views 0 copies

Python code

50 lines
Python 3.9+
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List


class SpanStatus(Enum):
    STARTED = "started"
    COMPLETED = "completed"


@dataclass
class SpanEvent:
    name: str
    timestamp: float = field(default_factory=time.time)
    attributes: dict = field(default_factory=dict)


class Span:
    def __init__(self, name: str):
        self.name = name
        self.events: List[SpanEvent] = []
        self._start_time = time.time()

    def add_event(self, name: str, attributes: dict = None) -> None:
        self.events.append(SpanEvent(name=name, attributes=attributes or {}))

    def mark_complete(self) -> None:
        self.add_event("span.completed", {"duration_ms": round((time.time() - self._start_time) * 1000, 2)})

    def to_dict(self) -> dict:
        return {
            "span": self.name,
            "events": [
                {
                    "name": e.name,
                    "ts": round(e.timestamp, 3),
                    "attrs": e.attributes
                }
                for e in self.events
            ]
        }


if __name__ == "__main__":
    span = Span("user_login")
    span.add_event("auth.attempted", {"method": "password"})
    span.add_event("auth.succeeded", {"user_id": 42})
    span.mark_complete()
    print(span.to_dict())

Output

stdout
{'span': 'user_login', 'events': [{'name': 'auth.attempted', 'ts': 1710000000.123, 'attrs': {'method': 'password'}}, {'name': 'auth.succeeded', 'ts': 1710000000.456, 'attrs': {'user_id': 42}}, {'name': 'span.completed', 'ts': 1710000000.789, 'attrs': {'duration_ms': 668.25}}]}

How it works

This code uses a dataclass to represent each SpanEvent with a default timestamp generated from time.time. The Span class keeps a list of events and uses add_event to append milestones; mark_complete adds a final event recording the duration in milliseconds. The to_dict method converts internal objects to a plain dictionary for easy JSON serialization. This pattern mirrors distributed tracing spans by capturing key moments with timestamps and attributes.

Common mistakes

  • Forgetting to use `field(default_factory=time.time)` with dataclasses to get a fresh timestamp per instance
  • Not rounding timestamps, leading to verbose floats in output
  • Assuming attributes are always provided; passing None requires a fallback
  • Not recording a start time before computing duration, causing negative or wrong values

Variations

  1. Use `time.perf_counter()` for high-resolution timing in benchmarks
  2. Add a context manager protocol to auto-mark completion on exit

Real-world use cases

  • Instrumenting a user login flow to track auth attempt and success milestones for latency analysis
  • Recording external API call events within an application span for distributed tracing
  • Capturing checkout lifecycle events in e-commerce to identify bottlenecks in payment processing

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Observability & SRE

Related tutorials and quizzes for this topic.