Timing

provide.foundation.utils.timing

TODO: Add module docstring.

Classes

Functions

timed_block

timed_block(
    logger_instance: FoundationLogger,
    event_name: str,
    layer_keys: dict[str, Any] | None = None,
    initial_kvs: dict[str, Any] | None = None,
    **extra_kvs: Any
) -> Generator[dict[str, Any], None, None]

Context manager that logs the duration of a code block.

Logs at DEBUG when entering, INFO on success, ERROR on exception.

Parameters:

Name	Type	Description	Default
logger_instance	FoundationLogger	Logger to use for output	required
event_name	str	Name of the operation being timed	required
layer_keys	dict[str, Any] | None	Semantic layer keys (e.g., llm-specific keys)	None
initial_kvs	dict[str, Any] | None	Initial key-value pairs to include in logs	None
**extra_kvs	Any	Additional key-value pairs	{}

Yields:

Type	Description
dict[str, Any]	A mutable dict that can be updated with additional context

Example

with timed_block(logger, "database_query") as ctx: ctx["query"] = "SELECT * FROM users" result = db.query("SELECT * FROM users") ctx["rows"] = len(result)

Source code in provide/foundation/utils/timing.py

@contextmanager
def timed_block(
    logger_instance: FoundationLogger,
    event_name: str,
    layer_keys: dict[str, Any] | None = None,
    initial_kvs: dict[str, Any] | None = None,
    **extra_kvs: Any,
) -> Generator[dict[str, Any], None, None]:
    """Context manager that logs the duration of a code block.

    Logs at DEBUG when entering, INFO on success, ERROR on exception.

    Args:
        logger_instance: Logger to use for output
        event_name: Name of the operation being timed
        layer_keys: Semantic layer keys (e.g., llm-specific keys)
        initial_kvs: Initial key-value pairs to include in logs
        **extra_kvs: Additional key-value pairs

    Yields:
        A mutable dict that can be updated with additional context

    Example:
        >>> with timed_block(logger, "database_query") as ctx:
        >>>     ctx["query"] = "SELECT * FROM users"
        >>>     result = db.query("SELECT * FROM users")
        >>>     ctx["rows"] = len(result)

    """
    # Combine all key-value pairs
    all_kvs = {}
    if layer_keys:
        all_kvs.update(layer_keys)
    if initial_kvs:
        all_kvs.update(initial_kvs)
    all_kvs.update(extra_kvs)

    # Try to get trace_id from context
    trace_id = _PROVIDE_CONTEXT_TRACE_ID.get()
    if trace_id and "trace_id" not in all_kvs:
        all_kvs["trace_id"] = trace_id

    # Create context dict that can be modified
    context: dict[str, Any] = {}

    # Log start
    logger_instance.debug(f"{event_name} started", **all_kvs)

    start_time = time.perf_counter()
    try:
        yield context

        # Success - calculate duration and log
        duration = time.perf_counter() - start_time
        all_kvs.update(context)
        all_kvs["duration_seconds"] = round(duration, 3)
        all_kvs["outcome"] = "success"

        logger_instance.info(f"{event_name} completed", **all_kvs)

    except Exception as e:
        # Error - calculate duration and log with exception
        duration = time.perf_counter() - start_time
        all_kvs.update(context)
        all_kvs["duration_seconds"] = round(duration, 3)
        all_kvs["outcome"] = "error"
        all_kvs["error.message"] = str(e)
        all_kvs["error.type"] = type(e).__name__

        logger_instance.error(f"{event_name} failed", exc_info=True, **all_kvs)
        raise