util

Python SDK reference for weave.trace.util

API Overview

---

class ContextAwareThread

A Thread that runs functions with the context of the caller.

This is a drop-in replacement for threading.Thread that ensures calls behave as expected inside the thread. Weave requires certain contextvars to be set (see call_context.py), but new threads do not automatically copy context from the parent, which can cause the call context to be lost — not good! This class automates contextvar copying so using this thread “just works” as the user probably expects.

You can achieve the same effect without this class by instead writing:

def run_with_context(func, *args, **kwargs):
     context = copy_context()
     def wrapper():
         context.run(func, *args, **kwargs)
     return wrapper

thread = threading.Thread(target=run_with_context(your_func, *args, **kwargs))
thread.start()

method __init__

__init__(*args: 'Any', **kwargs: 'Any') → None

---

property daemon

A boolean value indicating whether this thread is a daemon thread.

This must be set before start() is called, otherwise RuntimeError is raised. Its initial value is inherited from the creating thread; the main thread is not a daemon thread and therefore all threads created in the main thread default to daemon = False.

The entire Python program exits when only daemon threads are left.

---

property ident

Thread identifier of this thread or None if it has not been started.

This is a nonzero integer. See the get_ident() function. Thread identifiers may be recycled when a thread exits and another thread is created. The identifier is available even after the thread has exited.

---

property name

A string used for identification purposes only.

It has no semantics. Multiple threads may be given the same name. The initial name is set by the constructor.

---

property native_id

Native integral thread ID of this thread, or None if it has not been started.

This is a non-negative integer. See the get_native_id() function. This represents the Thread ID as reported by the kernel.

---

method run

run() → None

---

class ContextAwareThreadPoolExecutor

A ThreadPoolExecutor that runs functions with the context of the caller.

This is a drop-in replacement for concurrent.futures.ThreadPoolExecutor that ensures weave calls behave as expected inside the executor. Weave requires certain contextvars to be set (see call_context.py), but new threads do not automatically copy context from the parent, which can cause the call context to be lost — not good! This class automates contextvar copying so using this executor “just works” as the user probably expects.

You can achieve the same effect without this class by instead writing:

with concurrent.futures.ThreadPoolExecutor() as executor:
     contexts = [copy_context() for _ in range(len(vals))]

     def _wrapped_fn(*args):
         return contexts.pop().run(fn, *args)

     executor.map(_wrapped_fn, vals)

method __init__

__init__(*args: 'Any', **kwargs: 'Any') → None

---

method map

map(
    fn: 'Callable',
    *iterables: 'Iterable[Any]',
    timeout: 'float | None' = None,
    chunksize: 'int' = 1
) → Iterator

---

method submit

submit(fn: 'Callable', *args: 'Any', **kwargs: 'Any') → Any

---

class ContextAwareThread

A Thread that runs functions with the context of the caller.

This is a drop-in replacement for threading.Thread that ensures calls behave as expected inside the thread. Weave requires certain contextvars to be set (see call_context.py), but new threads do not automatically copy context from the parent, which can cause the call context to be lost — not good! This class automates contextvar copying so using this thread “just works” as the user probably expects.

You can achieve the same effect without this class by instead writing:

def run_with_context(func, *args, **kwargs):
     context = copy_context()
     def wrapper():
         context.run(func, *args, **kwargs)
     return wrapper

thread = threading.Thread(target=run_with_context(your_func, *args, **kwargs))
thread.start()

method __init__

__init__(*args: 'Any', **kwargs: 'Any') → None

---

property daemon

A boolean value indicating whether this thread is a daemon thread.

This must be set before start() is called, otherwise RuntimeError is raised. Its initial value is inherited from the creating thread; the main thread is not a daemon thread and therefore all threads created in the main thread default to daemon = False.

The entire Python program exits when only daemon threads are left.

---

property ident

Thread identifier of this thread or None if it has not been started.

This is a nonzero integer. See the get_ident() function. Thread identifiers may be recycled when a thread exits and another thread is created. The identifier is available even after the thread has exited.

---

property name

A string used for identification purposes only.

It has no semantics. Multiple threads may be given the same name. The initial name is set by the constructor.

---

property native_id

Native integral thread ID of this thread, or None if it has not been started.

This is a non-negative integer. See the get_native_id() function. This represents the Thread ID as reported by the kernel.

---

method run

run() → None

---

class ContextAwareThreadPoolExecutor

A ThreadPoolExecutor that runs functions with the context of the caller.

This is a drop-in replacement for concurrent.futures.ThreadPoolExecutor that ensures weave calls behave as expected inside the executor. Weave requires certain contextvars to be set (see call_context.py), but new threads do not automatically copy context from the parent, which can cause the call context to be lost — not good! This class automates contextvar copying so using this executor “just works” as the user probably expects.

You can achieve the same effect without this class by instead writing:

with concurrent.futures.ThreadPoolExecutor() as executor:
     contexts = [copy_context() for _ in range(len(vals))]

     def _wrapped_fn(*args):
         return contexts.pop().run(fn, *args)

     executor.map(_wrapped_fn, vals)

method __init__

__init__(*args: 'Any', **kwargs: 'Any') → None

---

method map

map(
    fn: 'Callable',
    *iterables: 'Iterable[Any]',
    timeout: 'float | None' = None,
    chunksize: 'int' = 1
) → Iterator

---

method submit

submit(fn: 'Callable', *args: 'Any', **kwargs: 'Any') → Any

---

function deprecated

deprecated(new_name: 'str') → Callable[[Callable[, Any]], Callable[, Any]]

Decorator to mark a function as deprecated and redirect users to new_name.

---

function is_colab

is_colab()

---

function is_notebook

is_notebook() → bool

---

function log_once

log_once(log_method: 'Callable[[str], None]', message: 'str') → None

Logs a message once, suppressing subsequent messages of the same type. This is useful for notifying the user about errors without spamming the logs.

This is mostly useful for cases where the same error message might occur many times. For example, if an op fails to save, it is likely going to happen every time that op is called. Or, if we have an error in our patched iterator, then it likely happens every time we iterate over the result. This allows use to inform the user about the error without clogging up their logs.

Args:

• log_method: The method to use to log the message. This should accept a string argument.
• message: The message to log. Example:

log_once(logger.error, "Failed to save op")