ddp_utils.call_tracer¶

import ddp_utils.call_tracer

Capture, format, and profile Python call information.

The module exposes a process-wide CallTracer singleton, decorators for recording calls and execution time, stack inspection helpers, and speedscope-compatible JSON export.

Examples

Record a timed call and inspect the slowest function:

from ddp_utils.call_tracer import hotspots, profile_time

@profile_time(label="load-record")
def load_record():
    return {"id": 7}

load_record()
slowest = hotspots(top_n=1)

class ddp_utils.call_tracer.OutputFormat(value)¶

Bases: Enum

Select how call records are rendered.

Examples

Configure tree output:

config = TraceConfig(output_format=OutputFormat.TREE)

class ddp_utils.call_tracer.TraceFilter(value)¶

Bases: Enum

Select the built-in frame filtering policy.

Examples

Retain every frame that passes explicit filters:

config = TraceConfig(filter_type=TraceFilter.ALL)

class ddp_utils.call_tracer.TraceConfig(output_format: OutputFormat = OutputFormat.SIMPLE, filter_type: TraceFilter = TraceFilter.EXCLUDE_LIBRARIES, max_depth: int = 20, show_line_numbers: bool = True, show_args: bool = False, show_return_values: bool = False, show_execution_time: bool = False, show_thread_info: bool = False, show_module_path: bool = False, include_modules: Set[str] = <factory>, exclude_modules: Set[str] = <factory>, include_functions: Set[str] = <factory>, exclude_functions: Set[str] = <factory>, colors: bool = True, indent_size: int = 2, separator: str = ' → ', log_to_file: str | None = None, log_level: str = 'INFO', skip_fast_calls: float = 0.0, sample_rate: float = 1.0)¶

Bases: object

Configure call collection, filtering, rendering, and persistence.

Variables

Name	Type	Description
output_format	OutputFormat	Renderer used for textual trace output.
filter_type	TraceFilter	Built-in frame filtering policy.
max_depth	int	Maximum number of records rendered by default.
show_line_numbers	bool	Whether detailed output includes source lines.
show_args	bool	Reserved switch for argument rendering.
show_return_values	bool	Reserved switch for return-value rendering.
show_execution_time	bool	Reserved switch for duration rendering.
show_thread_info	bool	Whether JSON output includes thread metadata.
show_module_path	bool	Whether tree output includes source filenames.
include_modules	Set[str]	Optional allowlist of source basenames.
exclude_modules	Set[str]	Source basenames to suppress.
include_functions	Set[str]	Optional allowlist of function names.
exclude_functions	Set[str]	Function names to suppress.
colors	bool	Whether detailed output uses ANSI colors.
indent_size	int	Spaces added per tree level.
separator	str	Separator used by simple and colored renderers.
log_to_file	str \| None	Optional path that receives printed active traces.
log_level	str	Reserved logging severity label.
skip_fast_calls	float	Reserved minimum duration threshold in seconds.
sample_rate	float	Reserved sampling ratio from `0.0` to `1.0`.

Examples

Create compact JSON trace output:

config = TraceConfig(
    output_format=OutputFormat.JSON,
    max_depth=10,
    show_thread_info=True,
)

class ddp_utils.call_tracer.CallInfo(filename: str, module: str, function: str, line_no: int, args: Dict[str, ~typing.Any]=<factory>, kwargs: Dict[str, ~typing.Any]=<factory>, return_value: Any = None, execution_time: float = 0.0, thread_id: int = 0, thread_name: str = '', timestamp: float = <factory>, call_id: str = '')¶

Bases: object

Store one captured function-call or timing record.

Variables

Name	Type	Description
filename	str	Source file basename.
module	str	Source path recorded by the tracer.
function	str	Qualified or frame function name.
line_no	int	Source line associated with the call.
args	Dict[str, Any]	Optional captured positional-argument mapping.
kwargs	Dict[str, Any]	Optional captured keyword arguments.
return_value	Any	Optional captured return value.
execution_time	float	Measured duration in seconds.
thread_id	int	Identifier of the executing thread.
thread_name	str	Name of the executing thread.
timestamp	float	Unix timestamp at record creation.
call_id	str	Trace-local record identifier.

Examples

Describe a synthetic call:

call = CallInfo(
    filename="worker.py",
    module="/app/worker.py",
    function="run",
    line_no=12,
)

class ddp_utils.call_tracer.CallTracer¶

Bases: object

Collect and render process-wide call and timing records.

CallTracer is a singleton. Reconstructing it returns the same object and preserves its configuration and accumulated records.

Examples

Capture decorated calls inside a named trace:

call_tracer = CallTracer()

@call_tracer.trace_call
def load():
    return "ready"

with call_tracer.trace("startup"):
    load()

Initialize singleton state exactly once.

Repeated construction leaves the existing configuration, trace history, and recorded calls unchanged.

Examples

Preserve configuration across repeated construction:

first = CallTracer().configure(max_depth=5)
assert CallTracer().config.max_depth == first.config.max_depth

configure(**kwargs) → CallTracer¶

Update recognized configuration fields in place.

Unknown names are ignored so callers may safely pass a wider settings mapping.

Parameters

Name	Description
**kwargs	Candidate `TraceConfig` field values.

Returns

Type	Description
CallTracer	This tracer instance for fluent configuration.

Examples

Select JSON output and a ten-record limit:

CallTracer().configure(
    output_format=OutputFormat.JSON,
    max_depth=10,
)

trace_call(func: Callable) → Callable¶

Decorate a synchronous callable and record every invocation.

The record is appended before the wrapped callable runs. Arguments and return values are not captured by this decorator.

Parameters

Name	Type	Description
func	Callable	Synchronous callable to wrap.

Returns

Type	Description
Callable	A metadata-preserving wrapper that records and invokes `func`.

Examples

Record calls to a worker function:

call_tracer = CallTracer()

@call_tracer.trace_call
def work(value):
    return value * 2

start_trace(trace_id: str | None = None) → str¶

Reset collected calls and begin a new trace session.

Parameters

Name	Type	Description
trace_id	str \| None	Optional stable identifier. A timestamped identifier is generated when omitted.

Returns

Type	Description
str	The active trace identifier.

Raises

Exception	Description
OSError	The configured log file cannot be opened for appending.

Examples

Start a named session:

trace_id = CallTracer().start_trace("import-job")

stop_trace() → List[CallInfo]¶

Stop the active trace and snapshot its collected records.

The snapshot is appended to the in-memory history and an open trace log is closed.

Returns

Type	Description
List[CallInfo]	A shallow copy of the current trace records.

Examples

Stop tracing and inspect the captured function names:

records = CallTracer().stop_trace()
names = [record.function for record in records]

trace(trace_id: str | None = None)¶

Run a trace session that is always stopped on context exit.

Parameters

Name	Type	Description
trace_id	str \| None	Optional identifier passed to `start_trace()`.

Yields

The active trace identifier.

Raises

Exception	Description
OSError	The configured trace log cannot be opened.

Examples

Bound collection to one operation:

with CallTracer().trace("batch") as trace_id:
    process_batch()

get_current_stack(skip_frames: int = 0, max_frames: int | None = None) → List[CallInfo]¶

Inspect the live Python stack and convert visible frames to calls.

Parameters

Name	Type	Description
skip_frames	int	Additional frames to omit after this method’s frame.
max_frames	int \| None	Maximum number of candidate frames to inspect. `None` inspects the remainder of the stack.

Returns

Type	Description
List[CallInfo]	Frame records that pass the configured filters.

Examples

Inspect at most five caller frames:

frames = CallTracer().get_current_stack(max_frames=5)

get_current_trace(max_depth: int | None = None) → str¶

Render recorded calls or, when inactive, the live Python stack.

Parameters

Name	Type	Description
max_depth	int \| None	Optional output record limit overriding the configured maximum depth.

Returns

Type	Description
str	Formatted trace text, or `"No calls to display"` when no visible live frames exist.

Examples

Render up to eight calls:

text = CallTracer().get_current_trace(max_depth=8)

print_stack(skip_frames: int = 0, max_frames: int | None = None, config: Dict[str, Any] | None = None, **kwargs) → None¶

Print the live call stack using optional temporary settings.

Parameters

Name	Type	Description
skip_frames	int	Additional caller frames to skip.
max_frames	int \| None	Maximum number of candidate frames to inspect.
config	Dict[str, Any] \| None	Temporary configuration values for this print operation.
**kwargs		Additional temporary `TraceConfig` values.

Examples

Print a short uncolored stack without changing persistent config:

CallTracer().print_stack(max_frames=3, colors=False)

print_trace(max_depth: int | None = None) → None¶

Print the current trace and append it to the open trace log.

Parameters

Name	Type	Description
max_depth	int \| None	Optional record limit for the rendered output.

Examples

Print the first ten records:

CallTracer().print_trace(max_depth=10)

profile_time(func: Callable | None = None, *, label: str | None = None) → Callable¶

Decorate a synchronous callable and record its execution time.

Timing is recorded in a finally block, including calls that raise. The decorator supports both bare and configured usage.

Parameters

Name	Type	Description
func	Callable \| None	Optional callable supplied by bare decorator syntax.
label	str \| None	Optional function label stored in the timing record.

Returns

Type	Description
Callable	A decorated callable when `func` is supplied; otherwise a decorator waiting for a callable.

Examples

Use the decorator with and without a custom label:

@tracer.profile_time
def parse():
    return "done"

@tracer.profile_time(label="database-query")
def fetch():
    return []

hotspots(top_n: int = 10) → List[Dict[str, Any]]¶

Aggregate timed records and return the slowest functions.

Parameters

Name	Type	Description
top_n	int	Maximum number of aggregate rows to return.

Returns

Type	Description
List[Dict[str, Any]]	Dictionaries containing `function`, `total_time`, `call_count`, `avg_time`, and `file`, ordered by descending total duration.

Examples

Read the three most expensive functions:

slowest = CallTracer().hotspots(top_n=3)

export_flamegraph(path: str) → None¶

Write collected timing records as speedscope evented JSON.

Parameters

Name	Type	Description
path	str	Destination JSON path. Its parent directory must exist.

Raises

Exception	Description
OSError	The destination cannot be opened or written.

Examples

Export a completed timing session for speedscope.app:

with tracer.trace("pipeline"):
    run_pipeline()
tracer.export_flamegraph("profile.json")

ddp_utils.call_tracer.trace_calls(config: Dict[str, Any] | None = None, as_decorator: bool = False, **kwargs) → Callable | CallTracer¶

Configure and return the shared tracer or its call decorator.

Parameters

Name	Type	Description
config	Dict[str, Any] \| None	Optional mapping of `TraceConfig` field values.
as_decorator	bool	Return a call-recording decorator when `True`.
**kwargs		Additional configuration values applied after `config`.

Returns

Type	Description
Callable \| CallTracer	The shared `CallTracer`, or a decorator backed by it.

Examples

Configure the shared tracer:

call_tracer = trace_calls(max_depth=8)

Decorate a function through the convenience API:

@trace_calls(as_decorator=True)
def load():
    return "ready"

ddp_utils.call_tracer.profile_time(func: Callable | None = None, *, label: str | None = None) → Callable¶

Decorate a callable with the shared tracer’s duration profiler.

Parameters

Name	Type	Description
func	Callable \| None	Optional callable supplied by bare decorator syntax.
label	str \| None	Optional label stored instead of the qualified function name.

Returns

Type	Description
Callable	A decorated callable or a decorator waiting for one.

Examples

Record a function under a stable label:

@profile_time(label="request")
def fetch():
    return "ok"

ddp_utils.call_tracer.hotspots(top_n: int = 10) → List[Dict[str, Any]]¶

Return slow-function aggregates from the shared tracer.

Parameters

Name	Type	Description
top_n	int	Maximum number of aggregate rows to return.

Returns

Type	Description
List[Dict[str, Any]]	Timing aggregates ordered by descending total duration.

Examples

Read the five most expensive functions:

slowest = hotspots(top_n=5)

ddp_utils.call_tracer.print_stack(skip_frames: int = 0, max_frames: int | None = None, **kwargs) → None¶

Print the live stack through the shared tracer.

Parameters

Name	Type	Description
skip_frames	int	Additional caller frames to omit.
max_frames	int \| None	Maximum number of candidate frames to inspect.
**kwargs		Temporary `TraceConfig` values for this output.

Examples

Print up to five frames as a tree:

print_stack(max_frames=5, output_format=OutputFormat.TREE)

ddp_utils.call_tracer.get_stack(skip_frames: int = 0, max_frames: int | None = None, **kwargs) → str¶

Render the live stack through the shared tracer.

Parameters

Name	Type	Description
skip_frames	int	Additional caller frames to omit.
max_frames	int \| None	Maximum number of candidate frames to inspect.
**kwargs		Temporary `TraceConfig` values for this rendering.

Returns

Type	Description
str	The formatted stack, or `"No calls recorded"` when no frame passes the active filters.

Examples

Produce stable uncolored detailed output:

text = get_stack(
    max_frames=5,
    output_format=OutputFormat.DETAILED,
    colors=False,
)