Repository: ray-project/ray
File: python/ray/data/_internal/util.py (Python)

Code before:
    return path


def _is_local_scheme(paths: Union[str, List[str]]) -> bool:
    """Returns True if the given paths are in local scheme.
    Note: The paths must be in same scheme, i.e. it's invalid and
    will raise error if paths are mixed with different schemes.
    """
    if isinstance(paths, str):
        paths = [paths]
    if isinstance(paths, pathlib.Path):
        paths = [str(paths)]
    elif not isinstance(paths, list) or any(not isinstance(p, str) for p in paths):
        raise ValueError("paths must be a path string or a list of path strings.")
    elif len(paths) == 0:
        raise ValueError("Must provide at least one path.")
    num = sum(urllib.parse.urlparse(path).scheme == _LOCAL_SCHEME for path in paths)
    if num > 0 and num < len(paths):
        raise ValueError(
            "The paths must all be local-scheme or not local-scheme, "
            f"but found mixed {paths}"
        )
    return num == len(paths)


def _validate_head_node_resources_for_local_scheduling(
    ray_remote_args: Dict[str, Any],
    *,
    op_name: str,
    required_num_cpus: int = 1,
    required_num_gpus: int = 0,
    required_memory: int = 0,
) -> None:
    """Ensure the head node has enough resources before pinning work there.

    Local paths (``local://``) and other driver-local I/O schedule tasks on the
    head node via ``NodeAffinitySchedulingStrategy``. If the head node was
    intentionally started with zero logical resources (a common practice to
    avoid OOMs), those tasks become unschedulable. Detect this upfront and
    raise a clear error with remediation steps.
    """

    # Ray defaults to reserving 1 CPU per task when num_cpus isn't provided.
    num_cpus = ray_remote_args.get("num_cpus", required_num_cpus)
    num_gpus = ray_remote_args.get("num_gpus", required_num_gpus)
    memory = ray_remote_args.get("memory", required_memory)

    # Resource keys follow the Resources map of ray.nodes() (e.g., CPU, GPU, memory).
    required_resources: Dict[str, float] = {}
    required_resources["CPU"] = float(num_cpus)
    required_resources["GPU"] = float(num_gpus)

Review comment (suggestion):
Nit: Consider moving this to `ray.data.datasource.util` if you think it makes more sense there than in the generic `util` file

Code after:
    return path


def _is_local_scheme(paths: Union[str, List[str]]) -> bool:
    """Returns True if the given paths are in local scheme.
    Note: The paths must be in same scheme, i.e. it's invalid and
    will raise error if paths are mixed with different schemes.
    """
    if isinstance(paths, str):
        paths = [paths]
    if isinstance(paths, pathlib.Path):
        paths = [str(paths)]
    elif not isinstance(paths, list) or any(not isinstance(p, str) for p in paths):
        raise ValueError("paths must be a path string or a list of path strings.")
    elif len(paths) == 0:
        raise ValueError("Must provide at least one path.")
    num = sum(urllib.parse.urlparse(path).scheme == _LOCAL_SCHEME for path in paths)
    if num > 0 and num < len(paths):
        raise ValueError(
            "The paths must all be local-scheme or not local-scheme, "
            f"but found mixed {paths}"
        )
    return num == len(paths)


def _truncated_repr(obj: Any) -> str:
    """Utility to return a truncated object representation for error messages."""
    msg = str(obj)
    if len(msg) > 200:
        msg = msg[:200] + "..."
    return msg


def _insert_doc_at_pattern(
    obj,
    *,
    message: str,
    pattern: str,
    insert_after: bool = True,
    directive: Optional[str] = None,
    skip_matches: int = 0,
) -> str:
    if "\n" in message:
        raise ValueError(
            "message shouldn't contain any newlines, since this function will insert "
            f"its own linebreaks when text wrapping: {message}"
        )

    doc = obj.__doc__.strip()
    if not doc:
        doc = ""