How to Model Span Events in Python
Define a Span class with timestamped milestone events and a completion marker to track operation lifecycle.
Python code
50 linesimport 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
{'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
- Use `time.perf_counter()` for high-resolution timing in benchmarks
- 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
More from Observability & SRE
- Adding a Correlation ID to Log Context in Python medium
- Calculate Error Rate from Log Stream in Python easy
- Check if a Timestamp Falls in a Daily Maintenance Window in Python easy
- Export Metrics with OTLP Mock in Python medium
- Generate Mock CPU and Memory Metrics in Python easy
- Generate Prometheus Text Exposition Format in Python easy
Keep learning
Related tutorials and quizzes for this topic.