nat.profiler.decorators.function_tracking#

Attributes#

F

Functions#

_serialize_data(→ Any)

Convert obj into a structure that can be passed to json.dumps(...).

_prepare_serialized_args_kwargs(→ tuple[list[Any], ...)

Serialize args and kwargs before calling the wrapped function.

push_intermediate_step(→ None)

Push an intermediate step to the NAT Event Stream.

track_function(…)

Decorator that can wrap any type of function (sync, async, generator,

track_unregistered_function(…)

Decorator that wraps any function with scope management and automatic tracking.

Module Contents#

_serialize_data(obj: Any) → Any#

Convert obj into a structure that can be passed to json.dumps(...).

_prepare_serialized_args_kwargs(
*args,
**kwargs,
) → tuple[list[Any], dict[str, Any]]#

Serialize args and kwargs before calling the wrapped function.

push_intermediate_step(
step_manager: nat.builder.intermediate_step_manager.IntermediateStepManager,
identifier: str,
function_name: str,
event_type: nat.data_models.intermediate_step.IntermediateStepType,
args: Any = None,
kwargs: Any = None,
output: Any = None,
metadata: dict[str, Any] | None = None,
) → None#

Push an intermediate step to the NAT Event Stream.

F#
track_function(
func: F,
*,
metadata: dict[str, Any] | None = None,
) → F#
track_function(
*,
metadata: dict[str, Any] | None = None,
) → collections.abc.Callable[[F], F]

Decorator that can wrap any type of function (sync, async, generator, async generator) and executes “tracking logic” around it.

  • If the function is async, it will be wrapped in an async function.

  • If the function is a generator, it will be wrapped in a generator function.

  • If the function is an async generator, it will be wrapped in an async generator function.

  • If the function is sync, it will be wrapped in a sync function.

track_unregistered_function(
func: F,
*,
name: str | None = None,
metadata: dict[str, Any] | None = None,
) → F#
track_unregistered_function(
*,
name: str | None = None,
metadata: dict[str, Any] | None = None,
) → collections.abc.Callable[[F], F]

Decorator that wraps any function with scope management and automatic tracking.

  • Sets active function context using the function name

  • Leverages Context.push_active_function for built-in tracking

  • Avoids duplicate tracking entries by relying on the library’s built-in systems

  • Supports sync/async functions and generators

Args:

func: The function to wrap (auto-detected when used without parentheses) name: Custom name to use for tracking instead of func.__name__ metadata: Additional metadata to include in tracking