Source code for runhouse.resources.folders.folder_factory

from pathlib import Path
from typing import Optional, Union

from runhouse.logger import get_logger

from runhouse.resources.folders.folder import Folder
from runhouse.resources.hardware.utils import _get_cluster_from

logger = get_logger(__name__)


[docs]def folder(
    name: Optional[str] = None,
    path: Optional[Union[str, Path]] = None,
    system: Optional[Union[str, "Cluster"]] = None,
    load_from_den: bool = True,
    dryrun: bool = False,
) -> Folder:
    """Creates a Runhouse folder object, which can be used to interact with the folder at the given path.

    Args:
        name (Optional[str]): Name to give the folder, to be re-used later on.
        path (Optional[str or Path]): Path (or path) that the folder is located at.
        system (Optional[str or Cluster]): File system or cluster name. If prpre-oviding a file system this must be one of:
            [``file``, ``s3``, ``gs``].
        load_from_den (bool): Whether to try loading the Folder resource from Den. (Default: ``True``)
        dryrun (bool): Whether to create the Folder if it doesn't exist, or load a Folder object as a dryrun.
            (Default: ``False``)

    Returns:
        Folder: The resulting folder.

    Example:
        >>> import runhouse as rh
        >>> rh.folder(name='training_imgs', path='remote_directory/images', system='s3').save()

        >>> # Load folder from above
        >>> reloaded_folder = rh.folder(name="training_imgs")
    """
    # TODO [DG] Include loud warning that relative paths are relative to the git root / working directory!

    if name and not any([path, system]):
        # If only the name is provided
        try:
            return Folder.from_name(name, load_from_den=load_from_den, dryrun=dryrun)
        except ValueError:
            # This is a rare instance where passing no constructor params is actually valid (anonymous folder),
            # so if we don't find the name, the user may still actually want to create a new folder.
            pass

    system = system or Folder.DEFAULT_FS
    if system == "s3":
        from .s3_folder import S3Folder

        logger.debug(f"Creating a S3 folder with name: {name}")
        return S3Folder(
            system=system,
            path=path,
            name=name,
            dryrun=dryrun,
        )
    elif system == "gs":
        from .gcs_folder import GCSFolder

        logger.debug(f"Creating a GS folder with name: {name}")
        return GCSFolder(
            system=system,
            path=path,
            name=name,
            dryrun=dryrun,
        )

    if system == Folder.DEFAULT_FS:
        logger.debug(f"Creating local folder with name: {name}")
        return Folder(
            system=system,
            path=path,
            name=name,
            dryrun=dryrun,
        )

    from runhouse import Cluster

    cluster_system = _get_cluster_from(system, dryrun=dryrun)
    if isinstance(cluster_system, Cluster):
        logger.debug(f"Creating folder {name} for cluster: {cluster_system.name}")
        return Folder(
            system=cluster_system,
            path=path,
            name=name,
            dryrun=dryrun,
        )

    raise ValueError(
        f"System '{system}' not currently supported. If the file system "
        f"is a cluster (ex: '/my-user/rh-cpu'), make sure the cluster config has been saved."
    )