Utilities

Helper functions and decorators.

Overview

The utils module provides:

• Validators: Input validation functions
• Decorators: Reusable function decorators

Validators

validate_dataset_identifier

Validates and resolves dataset identifiers:

from huwise_utils_py import validate_dataset_identifier

# Resolve a numeric dataset ID
resolved = validate_dataset_identifier(dataset_id="100123")

# Validation errors
validate_dataset_identifier()  # ValueError: neither provided
validate_dataset_identifier(dataset_id="100123", dataset_uid="da_abc")  # ValueError: both provided

Decorators

retry

Retry decorator with exponential backoff:

from huwise_utils_py.utils.decorators import retry
import httpx

@retry(httpx.HTTPError, tries=3, delay=1, backoff=2)
def fetch_data():
    response = httpx.get("https://api.example.com/data")
    response.raise_for_status()
    return response.json()

# First failure: waits 1s
# Second failure: waits 2s
# Third attempt: either succeeds or raises

Parameters:

• exceptions_to_check: Exception type(s) to catch
• tries: Number of attempts (default: 4)
• delay: Initial delay in seconds (default: 3)
• backoff: Multiplier for delay (default: 2)

API Reference

Validators for Huwise Utils.

This module provides validation functions for dataset identifiers and other input validation needs.

validate_dataset_identifier(dataset_id: str | None = None, dataset_uid: str | None = None, config: HuwiseConfig | None = None) -> str

Validate and resolve dataset identifier to UID.

Either dataset_id or dataset_uid must be provided, but not both. If dataset_id is provided, it will be resolved to the corresponding UID.

PARAMETER	DESCRIPTION
dataset_id	The numeric identifier of the dataset. TYPE: str | None DEFAULT: None
dataset_uid	The unique string identifier (UID) of the dataset. TYPE: str | None DEFAULT: None
config	Optional HuwiseConfig for ID resolution. TYPE: HuwiseConfig | None DEFAULT: None

RETURNS	DESCRIPTION
str	The dataset UID.

RAISES	DESCRIPTION
ValueError	If both dataset_id and dataset_uid are provided.
ValueError	If neither dataset_id nor dataset_uid is provided.
IndexError	If dataset_id cannot be resolved to a UID.

Examples:

Resolve a numeric ID to a UID:

uid = validate_dataset_identifier(dataset_id="12345")

Use a UID directly:

uid = validate_dataset_identifier(dataset_uid="da_abc123")

Source code in src/huwise_utils_py/utils/validators.py

def validate_dataset_identifier(
    dataset_id: str | None = None,
    dataset_uid: str | None = None,
    config: HuwiseConfig | None = None,
) -> str:
    """Validate and resolve dataset identifier to UID.

    Either dataset_id or dataset_uid must be provided, but not both.
    If dataset_id is provided, it will be resolved to the corresponding UID.

    Args:
        dataset_id: The numeric identifier of the dataset.
        dataset_uid: The unique string identifier (UID) of the dataset.
        config: Optional HuwiseConfig for ID resolution.

    Returns:
        The dataset UID.

    Raises:
        ValueError: If both dataset_id and dataset_uid are provided.
        ValueError: If neither dataset_id nor dataset_uid is provided.
        IndexError: If dataset_id cannot be resolved to a UID.

    Examples:
        Resolve a numeric ID to a UID:

        ```python
        uid = validate_dataset_identifier(dataset_id="12345")
        ```

        Use a UID directly:

        ```python
        uid = validate_dataset_identifier(dataset_uid="da_abc123")
        ```
    """
    if dataset_id is not None and dataset_uid is not None:
        raise ValueError("dataset_id and dataset_uid are mutually exclusive")

    if dataset_id is None and dataset_uid is None:
        raise ValueError("Either dataset_id or dataset_uid must be specified")

    if dataset_id is not None:
        config = config or HuwiseConfig.from_env()
        client = HttpClient(config)
        response = client.get("/datasets/", params={"dataset_id": dataset_id})
        return response.json()["results"][0]["uid"]

    return dataset_uid  # type: ignore[return-value]

Decorators for Huwise Utils.

This module provides reusable decorators following DCC coding standards.

retry(exceptions_to_check: type[Exception] | tuple[type[Exception], ...], tries: int = 4, delay: float = 3, backoff: float = 2) -> Any

Retry decorator with exponential backoff.

Retries the decorated function on specified exceptions with configurable attempts and exponential delay.

PARAMETER	DESCRIPTION
exceptions_to_check	Exception(s) to catch and retry on. TYPE: type[Exception] | tuple[type[Exception], ...]
tries	Number of attempts before giving up. TYPE: int DEFAULT: 4
delay	Initial delay between retries in seconds. TYPE: float DEFAULT: 3
backoff	Multiplier for delay after each retry. TYPE: float DEFAULT: 2

RETURNS	DESCRIPTION
Any	Decorated function with retry logic.

Example

@retry(ConnectionError, tries=3, delay=1)
def fetch_data():
    return requests.get("https://api.example.com")

Source code in src/huwise_utils_py/utils/decorators.py

def retry(
    exceptions_to_check: type[Exception] | tuple[type[Exception], ...],
    tries: int = 4,
    delay: float = 3,
    backoff: float = 2,
) -> Any:
    """Retry decorator with exponential backoff.

    Retries the decorated function on specified exceptions with
    configurable attempts and exponential delay.

    Args:
        exceptions_to_check: Exception(s) to catch and retry on.
        tries: Number of attempts before giving up.
        delay: Initial delay between retries in seconds.
        backoff: Multiplier for delay after each retry.

    Returns:
        Decorated function with retry logic.

    Example:
        ```python
        @retry(ConnectionError, tries=3, delay=1)
        def fetch_data():
            return requests.get("https://api.example.com")
        ```
    """

    def decorator(func: Any) -> Any:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return func(*args, **kwargs)
                except exceptions_to_check as e:
                    logger.warning(
                        "Function failed, retrying",
                        function=func.__name__,
                        error=str(e),
                        retries_left=mtries - 1,
                        delay_seconds=mdelay,
                    )
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            return func(*args, **kwargs)

        return wrapper

    return decorator

log_execution_time(func: Any) -> Any

Log the execution time of a function.

PARAMETER	DESCRIPTION
func	Function to decorate. TYPE: Any

RETURNS	DESCRIPTION
Any	Decorated function that logs execution time.

Source code in src/huwise_utils_py/utils/decorators.py

def log_execution_time(func: Any) -> Any:
    """Log the execution time of a function.

    Args:
        func: Function to decorate.

    Returns:
        Decorated function that logs execution time.
    """

    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        logger.info(
            "Function executed",
            function=func.__name__,
            duration_ms=round(elapsed * 1000, 2),
        )
        return result

    return wrapper