Source code for swarmauri.standard.tracing.concrete.SimpleTracer

from datetime import datetime
import uuid
from typing import Dict, Any, Optional

from swarmauri.core.tracing.ITracer import ITracer
from swarmauri.standard.tracing.concrete.SimpleTraceContext import SimpleTraceContext

[docs] class SimpleTracer(ITracer): _instance = None # Singleton instance placeholder
[docs] @classmethod def instance(cls): if cls._instance is None: cls._instance = cls() return cls._instance
def __init__(self): if SimpleTracer._instance is not None: raise RuntimeError("SimpleTracer is a singleton. Use SimpleTracer.instance().") self.trace_stack = []
[docs] def start_trace(self, name: str, initial_attributes: Optional[Dict[str, Any]] = None) -> SimpleTraceContext: trace_id = str(uuid.uuid4()) trace_context = SimpleTraceContext(trace_id, name, initial_attributes) self.trace_stack.append(trace_context) return trace_context
[docs] def end_trace(self): if self.trace_stack: completed_trace = self.trace_stack.pop() completed_trace.close() # Example of simply printing the completed trace; in practice, you might log it or store it elsewhere print(f"Trace Completed: {completed_trace.name}, Duration: {completed_trace.start_time} to {completed_trace.end_time}, Attributes: {completed_trace.attributes}")
[docs] def annotate_trace(self, key: str, value: Any): if not self.trace_stack: print("No active trace to annotate.") return current_trace = self.trace_stack[-1] current_trace.add_attribute(key, value)